This is an automated email from the ASF dual-hosted git repository.
pvillard31 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 a80a072c311 NIFI-16017: Do not allow node to Connect to cluster if it
has Connector in wrong Troubleshooting state (#11334)
a80a072c311 is described below
commit a80a072c311571f16d7fc3d1eaccc342ac200424
Author: Mark Payne <[email protected]>
AuthorDate: Fri Jun 12 23:25:28 2026 -0400
NIFI-16017: Do not allow node to Connect to cluster if it has Connector in
wrong Troubleshooting state (#11334)
---
.../serialization/VersionedFlowSynchronizer.java | 33 +++++
.../VersionedFlowSynchronizerTest.java | 86 ++++++++++++-
.../ClusteredConnectorTroubleshootingIT.java | 133 +++++++++++++++++++++
3 files changed, 249 insertions(+), 3 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java
index 2f0d7cb333e..29310984eb8 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizer.java
@@ -28,6 +28,7 @@ import org.apache.nifi.cluster.protocol.DataFlow;
import org.apache.nifi.cluster.protocol.StandardDataFlow;
import org.apache.nifi.components.connector.ConnectorNode;
import org.apache.nifi.components.connector.ConnectorRepository;
+import org.apache.nifi.components.connector.ConnectorState;
import org.apache.nifi.components.connector.ConnectorSyncMode;
import org.apache.nifi.components.connector.ConnectorSyncResult;
import org.apache.nifi.components.validation.ValidationStatus;
@@ -60,6 +61,7 @@ import org.apache.nifi.flow.ScheduledState;
import org.apache.nifi.flow.VersionedAsset;
import org.apache.nifi.flow.VersionedComponent;
import org.apache.nifi.flow.VersionedConnector;
+import org.apache.nifi.flow.VersionedConnectorState;
import org.apache.nifi.flow.VersionedControllerService;
import org.apache.nifi.flow.VersionedExternalFlow;
import org.apache.nifi.flow.VersionedFlowAnalysisRule;
@@ -223,6 +225,12 @@ public class VersionedFlowSynchronizer implements
FlowSynchronizer {
verifyNoConnectionsWithDataRemoved(existingDataFlow,
proposedFlow, controller, flowComparison);
}
+ // Ensure that no Connector is in Troubleshooting mode locally
while the proposed flow has it in a
+ // non-Troubleshooting state, or vice versa. Reconciling such a
mismatch would require either dropping the
+ // user's in-progress Troubleshooting flow or forcing the rest of
the cluster into Troubleshooting, so the
+ // node must not join until the mismatch is resolved.
+ verifyConnectorTroubleshootingStatesMatch(proposedFlow,
controller);
+
synchronizeFlow(controller, existingDataFlow, proposedFlow,
affectedComponents);
} finally {
if (!existingFlowEmpty) {
@@ -271,6 +279,31 @@ public class VersionedFlowSynchronizer implements
FlowSynchronizer {
}
}
+ private void verifyConnectorTroubleshootingStatesMatch(final DataFlow
proposedFlow, final FlowController controller) {
+ final VersionedDataflow proposedDataflow =
proposedFlow.getVersionedDataflow();
+ if (proposedDataflow == null || proposedDataflow.getConnectors() ==
null) {
+ return;
+ }
+
+ final ConnectorRepository connectorRepository =
controller.getConnectorRepository();
+ for (final VersionedConnector proposedConnector :
proposedDataflow.getConnectors()) {
+ final ConnectorNode localConnector =
connectorRepository.getConnector(proposedConnector.getInstanceIdentifier(),
ConnectorSyncMode.LOCAL_ONLY);
+ if (localConnector == null) {
+ continue;
+ }
+
+ final boolean localTroubleshooting =
localConnector.getCurrentState() == ConnectorState.TROUBLESHOOTING;
+ final boolean proposedTroubleshooting =
proposedConnector.getScheduledState() ==
VersionedConnectorState.TROUBLESHOOTING;
+ if (localTroubleshooting != proposedTroubleshooting) {
+ final String troubleshootingDescription = localTroubleshooting
+ ? "in Troubleshooting mode on this node but is not in
Troubleshooting mode in the cluster flow"
+ : "not in Troubleshooting mode on this node but is in
Troubleshooting mode in the cluster flow";
+ throw new UninheritableFlowException("Proposed flow is not
inheritable by the flow controller because " + localConnector + " is " +
troubleshootingDescription
+ + ". Exit or enter Troubleshooting mode for this Connector
so that its state matches the cluster before joining.");
+ }
+ }
+ }
+
private void mapCompatibleBundles(final DataFlow proposedFlow, final
ExtensionManager extensionManager) {
final Set<String> missingComponentIds =
proposedFlow.getMissingComponents();
final VersionedDataflow dataflow = proposedFlow.getVersionedDataflow();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java
index 91ea1a87471..96a1432301f 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/serialization/VersionedFlowSynchronizerTest.java
@@ -20,6 +20,7 @@ import org.apache.nifi.cluster.protocol.DataFlow;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.connector.ConnectorNode;
import org.apache.nifi.components.connector.ConnectorRepository;
+import org.apache.nifi.components.connector.ConnectorState;
import org.apache.nifi.components.connector.ConnectorSyncMode;
import org.apache.nifi.components.connector.ConnectorSyncResult;
import org.apache.nifi.controller.FlowController;
@@ -53,6 +54,8 @@ import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
import java.io.File;
import java.nio.charset.StandardCharsets;
@@ -64,6 +67,7 @@ import java.util.UUID;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.inOrder;
@@ -73,6 +77,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
class VersionedFlowSynchronizerTest {
private static final String FLOW_CONFIGURATION = "flow.json.gz";
@@ -367,7 +372,7 @@ class VersionedFlowSynchronizerTest {
proposedConnector.setScheduledState(VersionedConnectorState.ENABLED);
final ConnectorNode orphanConnector = mock(ConnectorNode.class);
-
org.mockito.Mockito.lenient().when(orphanConnector.getIdentifier()).thenReturn("orphan-connector-id");
+
when(orphanConnector.getIdentifier()).thenReturn("orphan-connector-id");
when(connectorRepository.stopConnector(orphanConnector))
.thenReturn(java.util.concurrent.CompletableFuture.completedFuture(null));
@@ -396,7 +401,7 @@ class VersionedFlowSynchronizerTest {
final ConnectorRepository connectorRepository =
mock(ConnectorRepository.class);
final ConnectorNode existingConnector = mock(ConnectorNode.class);
-
org.mockito.Mockito.lenient().when(existingConnector.getIdentifier()).thenReturn(connectorId);
+ when(existingConnector.getIdentifier()).thenReturn(connectorId);
final VersionedConnector versionedConnector = new VersionedConnector();
versionedConnector.setInstanceIdentifier(connectorId);
@@ -436,7 +441,7 @@ class VersionedFlowSynchronizerTest {
proposedConnector.setScheduledState(VersionedConnectorState.ENABLED);
final ConnectorNode orphanConnector = mock(ConnectorNode.class);
-
org.mockito.Mockito.lenient().when(orphanConnector.getIdentifier()).thenReturn("orphan-connector-id");
+
when(orphanConnector.getIdentifier()).thenReturn("orphan-connector-id");
final java.util.concurrent.CompletableFuture<Void> failedFuture = new
java.util.concurrent.CompletableFuture<>();
failedFuture.completeExceptionally(new RuntimeException("Stop
failed"));
@@ -457,4 +462,79 @@ class VersionedFlowSynchronizerTest {
verify(connectorRepository).syncConnector(proposedConnector);
verify(orphanConnector).markInvalid(eq("Flow Synchronization
Failure"), any());
}
+
+ @Test
+ void testSyncFailsWhenLocalConnectorTroubleshootingButClusterIsNot() {
+ setRootGroup();
+
+ final String connectorId = UUID.randomUUID().toString();
+ final VersionedConnector proposedConnector =
createVersionedConnector(connectorId, VersionedConnectorState.RUNNING);
+
+ final ConnectorRepository connectorRepository =
mock(ConnectorRepository.class);
+ final ConnectorNode localConnector = mock(ConnectorNode.class);
+
when(localConnector.getCurrentState()).thenReturn(ConnectorState.TROUBLESHOOTING);
+ when(connectorRepository.getConnector(connectorId,
ConnectorSyncMode.LOCAL_ONLY)).thenReturn(localConnector);
+
+ setFlowController(connectorRepository);
+
when(versionedDataflow.getConnectors()).thenReturn(List.of(proposedConnector));
+
+ final UninheritableFlowException thrown =
assertThrows(UninheritableFlowException.class, () ->
+ versionedFlowSynchronizer.sync(flowController, dataFlow,
flowService, BundleUpdateStrategy.USE_SPECIFIED_OR_GHOST));
+ assertTrue(thrown.getMessage().contains("Troubleshooting"));
+ verify(connectorRepository, never()).syncConnector(any());
+ }
+
+ @Test
+ void testSyncFailsWhenClusterConnectorTroubleshootingButLocalIsNot() {
+ setRootGroup();
+
+ final String connectorId = UUID.randomUUID().toString();
+ final VersionedConnector proposedConnector =
createVersionedConnector(connectorId, VersionedConnectorState.TROUBLESHOOTING);
+
+ final ConnectorRepository connectorRepository =
mock(ConnectorRepository.class);
+ final ConnectorNode localConnector = mock(ConnectorNode.class);
+
when(localConnector.getCurrentState()).thenReturn(ConnectorState.RUNNING);
+ when(connectorRepository.getConnector(connectorId,
ConnectorSyncMode.LOCAL_ONLY)).thenReturn(localConnector);
+
+ setFlowController(connectorRepository);
+
when(versionedDataflow.getConnectors()).thenReturn(List.of(proposedConnector));
+
+ assertThrows(UninheritableFlowException.class, () ->
+ versionedFlowSynchronizer.sync(flowController, dataFlow,
flowService, BundleUpdateStrategy.USE_SPECIFIED_OR_GHOST));
+ verify(connectorRepository, never()).syncConnector(any());
+ }
+
+ @Test
+ void testSyncProceedsWhenConnectorTroubleshootingStateMatchesCluster() {
+ setRootGroup();
+
+ final String connectorId = UUID.randomUUID().toString();
+ final VersionedConnector proposedConnector =
createVersionedConnector(connectorId, VersionedConnectorState.TROUBLESHOOTING);
+
+ final ConnectorRepository connectorRepository =
mock(ConnectorRepository.class);
+ final ConnectorNode localConnector = mock(ConnectorNode.class);
+
when(localConnector.getCurrentState()).thenReturn(ConnectorState.TROUBLESHOOTING);
+ when(connectorRepository.getConnector(connectorId,
ConnectorSyncMode.LOCAL_ONLY)).thenReturn(localConnector);
+
+ final ConnectorNode syncedNode = mock(ConnectorNode.class);
+ when(connectorRepository.syncConnector(proposedConnector))
+
.thenReturn(ConnectorSyncResult.syncedConfigUnchanged(syncedNode,
VersionedConnectorState.TROUBLESHOOTING));
+
+ setFlowController(connectorRepository);
+
when(versionedDataflow.getConnectors()).thenReturn(List.of(proposedConnector));
+
+ versionedFlowSynchronizer.sync(flowController, dataFlow, flowService,
BundleUpdateStrategy.USE_SPECIFIED_OR_GHOST);
+
+ verify(connectorRepository).syncConnector(proposedConnector);
+ }
+
+ private VersionedConnector createVersionedConnector(final String
connectorId, final VersionedConnectorState scheduledState) {
+ final VersionedConnector versionedConnector = new VersionedConnector();
+ versionedConnector.setInstanceIdentifier(connectorId);
+ versionedConnector.setName("Test Connector");
+ versionedConnector.setType("org.apache.nifi.connectors.TestConnector");
+ versionedConnector.setBundle(CORE_BUNDLE);
+ versionedConnector.setScheduledState(scheduledState);
+ return versionedConnector;
+ }
}
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorTroubleshootingIT.java
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorTroubleshootingIT.java
index 99d50f0bf06..e7a7ac850be 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorTroubleshootingIT.java
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/java/org/apache/nifi/tests/system/connectors/ClusteredConnectorTroubleshootingIT.java
@@ -17,6 +17,7 @@
package org.apache.nifi.tests.system.connectors;
+import org.apache.nifi.cluster.coordination.node.NodeConnectionState;
import org.apache.nifi.components.connector.ConnectorState;
import org.apache.nifi.tests.system.NiFiInstanceFactory;
import org.apache.nifi.toolkit.client.NiFiClientException;
@@ -76,6 +77,138 @@ public class ClusteredConnectorTroubleshootingIT extends
ConnectorTroubleshootin
assertConnectorState(connectorId, ConnectorState.STOPPED);
}
+ /**
+ * Verifies that a node refuses to rejoin the cluster when it holds a
RUNNING Connector in Troubleshooting locally
+ * while the cluster's authoritative state for that Connector is not
Troubleshooting.
+ *
+ * <p>The sequence is:</p>
+ * <ol>
+ * <li>Start the Connector so the cluster's authoritative state for it
is RUNNING.</li>
+ * <li>Disconnect Node 2.</li>
+ * <li>Issue a node-local request against the disconnected Node 2 to
enter Troubleshooting, so only Node 2 is in
+ * Troubleshooting while the connected coordinator (Node 1) keeps
the Connector RUNNING.</li>
+ * <li>Reconnect Node 2.</li>
+ * </ol>
+ *
+ * <p>Node 2 has the Connector in Troubleshooting locally but the
cluster's authoritative state is not Troubleshooting,
+ * so the troubleshooting-state mismatch makes the proposed cluster flow
uninheritable. Node 2 returns to DISCONNECTED
+ * and keeps its own local Troubleshooting flow rather than adopting the
cluster's flow.</p>
+ */
+ @Test
+ public void
testNodeCannotRejoinWhenRunningConnectorWentTroubleshootingWhileDisconnected()
+ throws NiFiClientException, IOException, InterruptedException {
+
+ final ConnectorEntity connector =
getClientUtil().createConnector("ComponentLifecycleConnector");
+ final String connectorId = connector.getId();
+
+ getClientUtil().applyConnectorUpdate(connector);
+ getClientUtil().waitForValidConnector(connectorId);
+
+ getClientUtil().startConnector(connectorId);
+ assertConnectorState(connectorId, ConnectorState.RUNNING);
+
+ disconnectNode(2);
+ enterTroubleshootingOnDisconnectedNode2(connectorId);
+
+ reconnectNode(2);
+
+ assertNode2FailsToRejoinAndKeepsLocalState(connectorId,
ConnectorState.TROUBLESHOOTING);
+ }
+
+ /**
+ * Companion to {@link
#testNodeCannotRejoinWhenRunningConnectorWentTroubleshootingWhileDisconnected()}
for a STOPPED
+ * Connector. The cluster's authoritative state is STOPPED while Node 2 is
locally in Troubleshooting. As with the
+ * RUNNING case, the troubleshooting-state mismatch makes the cluster flow
uninheritable, so Node 2 must refuse to
+ * join and remain DISCONNECTED with its local Troubleshooting flow intact.
+ */
+ @Test
+ public void
testNodeCannotRejoinWhenStoppedConnectorWentTroubleshootingWhileDisconnected()
+ throws NiFiClientException, IOException, InterruptedException {
+
+ final ConnectorEntity connector =
getClientUtil().createConnector("ComponentLifecycleConnector");
+ final String connectorId = connector.getId();
+
+ getClientUtil().applyConnectorUpdate(connector);
+ getClientUtil().waitForValidConnector(connectorId);
+ assertConnectorState(connectorId, ConnectorState.STOPPED);
+
+ disconnectNode(2);
+ enterTroubleshootingOnDisconnectedNode2(connectorId);
+
+ reconnectNode(2);
+
+ assertNode2FailsToRejoinAndKeepsLocalState(connectorId,
ConnectorState.TROUBLESHOOTING);
+ }
+
+ /**
+ * Companion to {@link
#testNodeCannotRejoinWhenStoppedConnectorWentTroubleshootingWhileDisconnected()}
for the
+ * reverse mismatch: Node 2 is not in Troubleshooting locally while the
cluster's authoritative state is
+ * Troubleshooting.
+ *
+ * <p>The Connector is left STOPPED, then Node 2 is disconnected and the
Connector is taken into Troubleshooting on the
+ * remaining coordinator, so the cluster's authoritative state is
Troubleshooting while Node 2 stays STOPPED locally.
+ * When Node 2 reconnects, its local non-Troubleshooting state conflicts
with the cluster's Troubleshooting state, so
+ * the cluster flow is uninheritable and Node 2 must refuse to join,
remaining DISCONNECTED with its local STOPPED flow
+ * intact.</p>
+ */
+ @Test
+ public void
testNodeCannotRejoinWhenClusterEnteredTroubleshootingWhileNodeDisconnected()
+ throws NiFiClientException, IOException, InterruptedException {
+
+ final ConnectorEntity connector =
getClientUtil().createConnector("ComponentLifecycleConnector");
+ final String connectorId = connector.getId();
+
+ getClientUtil().applyConnectorUpdate(connector);
+ getClientUtil().waitForValidConnector(connectorId);
+ assertConnectorState(connectorId, ConnectorState.STOPPED);
+
+ disconnectNode(2);
+
+ getClientUtil().enterTroubleshooting(connectorId);
+ assertConnectorState(connectorId, ConnectorState.TROUBLESHOOTING);
+
+ reconnectNode(2);
+
+ assertNode2FailsToRejoinAndKeepsLocalState(connectorId,
ConnectorState.STOPPED);
+ }
+
+ private void assertNode2FailsToRejoinAndKeepsLocalState(final String
connectorId, final ConnectorState expectedNode2State)
+ throws InterruptedException, NiFiClientException, IOException {
+ // The reconnect must be rejected because Node 2's local
Troubleshooting state does not match the cluster, so Node 2
+ // returns to DISCONNECTED rather than reaching CONNECTED.
+ waitForNodeState(2, NodeConnectionState.DISCONNECTED);
+
+ // Only the coordinator remains connected.
+ assertEquals(1,
getNifiClient().getFlowClient().getClusterSummary().getClusterSummary().getConnectedNodeCount().intValue());
+
+ // Node 2 kept its own local flow rather than adopting the cluster's
conflicting flow.
+ waitForNode2ConnectorState(connectorId, expectedNode2State);
+ }
+
+ private void enterTroubleshootingOnDisconnectedNode2(final String
connectorId) throws NiFiClientException, IOException, InterruptedException {
+ switchClientToNode(2);
+ try {
+ final ConnectorEntity node2Connector =
getNifiClient().getConnectorClient(DO_NOT_REPLICATE).getConnector(connectorId);
+ node2Connector.setDisconnectedNodeAcknowledged(true);
+
getNifiClient().getConnectorClient(DO_NOT_REPLICATE).enterTroubleshooting(node2Connector);
+
+ waitFor(() -> ConnectorState.TROUBLESHOOTING.name().equals(
+
getNifiClient().getConnectorClient(DO_NOT_REPLICATE).getConnector(connectorId).getComponent().getState()));
+ } finally {
+ switchClientToNode(1);
+ }
+ }
+
+ private void waitForNode2ConnectorState(final String connectorId, final
ConnectorState expected) throws InterruptedException {
+ switchClientToNode(2);
+ try {
+ waitFor(() -> expected.name().equals(
+
getNifiClient().getConnectorClient(DO_NOT_REPLICATE).getConnector(connectorId).getComponent().getState()));
+ } finally {
+ switchClientToNode(1);
+ }
+ }
+
private void assertConnectorState(final String connectorId, final
ConnectorState expected) throws NiFiClientException, IOException {
final ConnectorEntity entity =
getNifiClient().getConnectorClient().getConnector(connectorId);
assertEquals(expected.name(), entity.getComponent().getState());
