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 42c3006b123 NIFI-16006: Addressed issue where a node can be
disconnected, then so… (#11331)
42c3006b123 is described below
commit 42c3006b12304d309ee573c3069c9f1ff4806920
Author: Mark Payne <[email protected]>
AuthorDate: Fri Jun 12 11:59:28 2026 -0400
NIFI-16006: Addressed issue where a node can be disconnected, then so…
(#11331)
* NIFI-16006: Addressed issue where a node can be disconnected, then soon
after re-connect but then be quickly told to disconnect due to a queued up
'Disconnect' message from the original disconnection. Now, we use a
'generation' flag so we know to ignore the message, and we also cancel the
background task that is trying to deliver it.
---
.../coordination/node/NodeClusterCoordinator.java | 82 ++++++++-----
.../node/TestNodeClusterCoordinator.java | 27 +++++
.../nifi/controller/StandardFlowService.java | 37 ++++--
.../nifi/controller/TestStandardFlowService.java | 135 +++++++++++++++++++++
4 files changed, 241 insertions(+), 40 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/NodeClusterCoordinator.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/NodeClusterCoordinator.java
index 16e1e17c7fa..e811fb95336 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/NodeClusterCoordinator.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/NodeClusterCoordinator.java
@@ -128,6 +128,7 @@ public class NodeClusterCoordinator implements
ClusterCoordinator, ProtocolHandl
private final ConcurrentMap<String, NodeConnectionStatus> nodeStatuses =
new ConcurrentHashMap<>();
private final ConcurrentMap<String, CircularFifoQueue<NodeEvent>>
nodeEvents = new ConcurrentHashMap<>(); //NOPMD
+ private final ConcurrentMap<String, Thread> pendingDisconnections = new
ConcurrentHashMap<>();
private final List<ClusterTopologyEventListener> eventListeners = new
CopyOnWriteArrayList<>();
@@ -1006,47 +1007,57 @@ public class NodeClusterCoordinator implements
ClusterCoordinator, ProtocolHandl
private Future<Void> disconnectAsynchronously(final DisconnectMessage
request, final int attempts, final int retrySeconds) {
final CompletableFuture<Void> future = new CompletableFuture<>();
+ final String nodeUuid = request.getNodeId().getId();
final Thread disconnectThread = new Thread(() -> {
final NodeIdentifier nodeId = request.getNodeId();
- Exception lastException = null;
- for (int i = 0; i < attempts; i++) {
- // If the node is restarted, it will attempt to reconnect. In
that case, we don't want to disconnect the node
- // again. So we instead log the fact that the state has now
transitioned to this point and consider the task completed.
- final NodeConnectionState currentConnectionState =
getConnectionState(nodeId);
- if (currentConnectionState == NodeConnectionState.CONNECTING
|| currentConnectionState == NodeConnectionState.CONNECTED) {
- reportEvent(nodeId, Severity.INFO, String.format(
- "State of Node %s has now transitioned from
DISCONNECTED to %s so will no longer attempt to notify node that it is
disconnected.", nodeId, currentConnectionState));
- future.completeExceptionally(new
IllegalStateException("Node was marked as disconnected but its state
transitioned from DISCONNECTED back to " + currentConnectionState +
- " before the node could be notified. This typically
indicates that the node was restarted."));
-
- return;
- }
+ try {
+ for (int i = 0; i < attempts; i++) {
+ if (Thread.currentThread().isInterrupted()) {
+ logger.info("Disconnect request for {} was cancelled
because the node submitted a new Connection Request", nodeId);
+ future.cancel(true);
+ return;
+ }
- // Try to send disconnect notice to the node
- try {
- senderListener.disconnect(request);
- reportEvent(nodeId, Severity.INFO, "Node disconnected due
to " + request.getExplanation());
- future.complete(null);
- return;
- } catch (final Exception e) {
- logger.error("Failed to notify {} that it has been
disconnected from the cluster due to {}", request.getNodeId(),
request.getExplanation());
- lastException = e;
+ final NodeConnectionState currentConnectionState =
getConnectionState(nodeId);
+ if (currentConnectionState ==
NodeConnectionState.CONNECTING || currentConnectionState ==
NodeConnectionState.CONNECTED) {
+ reportEvent(nodeId, Severity.INFO, String.format(
+ "State of Node %s has now transitioned from
DISCONNECTED to %s so will no longer attempt to notify node that it is
disconnected.", nodeId, currentConnectionState));
+ future.completeExceptionally(new
IllegalStateException("Node was marked as disconnected but its state
transitioned from DISCONNECTED back to " + currentConnectionState
+ + " before the node could be notified. This
typically indicates that the node was restarted."));
+ return;
+ }
try {
- Thread.sleep(retrySeconds * 1000L);
- } catch (final InterruptedException ie) {
- future.completeExceptionally(ie);
- Thread.currentThread().interrupt();
+ senderListener.disconnect(request);
+ reportEvent(nodeId, Severity.INFO, "Node disconnected
due to " + request.getExplanation());
+ future.complete(null);
return;
+ } catch (final Exception e) {
+ logger.error("Failed to notify {} that it has been
disconnected from the cluster due to {}", request.getNodeId(),
request.getExplanation());
+
+ try {
+ Thread.sleep(retrySeconds * 1000L);
+ } catch (final InterruptedException ie) {
+ logger.info("Disconnect request for {} was
cancelled because the node submitted a new Connection Request", nodeId);
+ future.cancel(true);
+ Thread.currentThread().interrupt();
+ return;
+ }
}
}
- }
- future.completeExceptionally(lastException);
+ future.completeExceptionally(new ProtocolException("Failed to
disconnect " + request.getNodeId() + " after " + attempts + " attempts"));
+ } finally {
+ pendingDisconnections.remove(nodeUuid, Thread.currentThread());
+ }
}, "Disconnect " + request.getNodeId());
+ final Thread previousThread = pendingDisconnections.put(nodeUuid,
disconnectThread);
+ if (previousThread != null) {
+ previousThread.interrupt();
+ }
disconnectThread.start();
return future;
}
@@ -1270,6 +1281,8 @@ public class NodeClusterCoordinator implements
ClusterCoordinator, ProtocolHandl
final DataFlow dataFlow =
requestMessage.getConnectionRequest().getDataFlow();
final ConnectionRequest requestWithNodeIdentities = new
ConnectionRequest(withNodeIdentities, dataFlow);
+ cancelPendingDisconnection(nodeIdentifier);
+
// Resolve Node identifier.
registerNodeId(nodeIdentifier);
@@ -1297,6 +1310,19 @@ public class NodeClusterCoordinator implements
ClusterCoordinator, ProtocolHandl
return createConnectionResponse(requestWithNodeIdentities,
nodeIdentifier);
}
+ private void cancelPendingDisconnection(final NodeIdentifier
nodeIdentifier) {
+ final Thread disconnectThread =
pendingDisconnections.remove(nodeIdentifier.getId());
+ if (disconnectThread != null) {
+ disconnectThread.interrupt();
+ logger.info("Received Connection Request from {} while a
disconnect request was still pending; interrupted the disconnect thread",
nodeIdentifier);
+ }
+ }
+
+ // Visible for testing
+ Thread getPendingDisconnectionThread(final NodeIdentifier nodeIdentifier) {
+ return pendingDisconnections.get(nodeIdentifier.getId());
+ }
+
private ConnectionResponseMessage createFlowElectionInProgressResponse() {
final ConnectionResponseMessage responseMessage = new
ConnectionResponseMessage();
final String statusDescription = flowElection.getStatusDescription();
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/node/TestNodeClusterCoordinator.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/node/TestNodeClusterCoordinator.java
index ad4741c33fb..8b03e1d57c6 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/node/TestNodeClusterCoordinator.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/node/TestNodeClusterCoordinator.java
@@ -22,6 +22,7 @@ import org.apache.nifi.cluster.protocol.ConnectionRequest;
import org.apache.nifi.cluster.protocol.ConnectionResponse;
import org.apache.nifi.cluster.protocol.DataFlow;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.cluster.protocol.ProtocolException;
import org.apache.nifi.cluster.protocol.StandardDataFlow;
import
org.apache.nifi.cluster.protocol.impl.ClusterCoordinationProtocolSenderListener;
import org.apache.nifi.cluster.protocol.message.ConnectionRequestMessage;
@@ -493,6 +494,32 @@ public class TestNodeClusterCoordinator {
assertEquals(4848, registeredId.getLoadBalancePort());
}
+ @Test
+ @Timeout(value = 10)
+ public void testConnectionRequestCancelsPendingDisconnect() throws
InterruptedException {
+ final NodeIdentifier nodeId1 = createNodeId(1);
+ final NodeIdentifier nodeId2 = createNodeId(2);
+ coordinator.updateNodeStatus(new NodeConnectionStatus(nodeId1,
NodeConnectionState.CONNECTED));
+ coordinator.updateNodeStatus(new NodeConnectionStatus(nodeId2,
NodeConnectionState.CONNECTED));
+
+ while (nodeStatuses.size() < 2) {
+ Thread.sleep(10L);
+ }
+ nodeStatuses.clear();
+
+ Mockito.doThrow(new
ProtocolException("Simulated")).when(senderListener).disconnect(any());
+ coordinator.requestNodeDisconnect(nodeId2,
DisconnectionCode.USER_DISCONNECTED, "Unit Test");
+
+ final Thread disconnectThread =
coordinator.getPendingDisconnectionThread(nodeId2);
+ assertNotNull(disconnectThread);
+ assertTrue(disconnectThread.isAlive());
+
+ requestConnection(nodeId2, coordinator);
+
+ assertNull(coordinator.getPendingDisconnectionThread(nodeId2));
+ assertTrue(disconnectThread.isInterrupted());
+ }
+
private NodeIdentifier createNodeId(final int index) {
return new NodeIdentifier(String.valueOf(index), "localhost", 8000 +
index, "localhost", 9000 + index, "localhost", 10000 + index, 11000 + index,
false);
}
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
index fc212ad88e4..1294525bceb 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
@@ -92,6 +92,7 @@ import java.util.UUID;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -119,6 +120,7 @@ public class StandardFlowService implements FlowService,
ProtocolHandler {
private final AtomicBoolean running = new AtomicBoolean(false);
private final AtomicReference<ScheduledExecutorService> executor = new
AtomicReference<>(null);
private final AtomicReference<SaveHolder> saveHolder = new
AtomicReference<>(null);
+ private final AtomicLong connectionGeneration = new AtomicLong(0);
private final ClusterCoordinator clusterCoordinator;
private final RevisionManager revisionManager;
private final NarManager narManager;
@@ -426,7 +428,8 @@ public class StandardFlowService implements FlowService,
ProtocolHandler {
return null;
}
case DISCONNECTION_REQUEST: {
- final Thread t = new Thread(() ->
handleDisconnectionRequest((DisconnectMessage) request), "Disconnect from
Cluster");
+ final long generationAtDisconnect =
connectionGeneration.get();
+ final Thread t = new Thread(() ->
handleDisconnectionRequest((DisconnectMessage) request,
generationAtDisconnect), "Disconnect from Cluster");
t.setDaemon(true);
t.start();
@@ -600,6 +603,11 @@ public class StandardFlowService implements FlowService,
ProtocolHandler {
}
}
+ // Visible for testing
+ long getConnectionGeneration() {
+ return connectionGeneration.get();
+ }
+
private void handleReconnectionRequest(final ReconnectionRequestMessage
request) {
try {
logger.info("Processing reconnection request from cluster
coordinator.");
@@ -719,32 +727,35 @@ public class StandardFlowService implements FlowService,
ProtocolHandler {
}
}
- private void handleDisconnectionRequest(final DisconnectMessage request) {
+ // Visible for testing
+ void handleDisconnectionRequest(final DisconnectMessage request, final
long expectedConnectionGeneration) {
logger.info("Received disconnection request message from cluster
coordinator with explanation: {}", request.getExplanation());
- disconnect(request.getExplanation());
+
+ writeLock.lock();
+ try {
+ if (connectionGeneration.get() != expectedConnectionGeneration) {
+ logger.info("Ignoring disconnection request [{}] because the
node has reconnected since the request was issued", request.getExplanation());
+ return;
+ }
+
+ disconnect(request.getExplanation());
+ } finally {
+ writeLock.unlock();
+ }
}
private void disconnect(final String explanation) {
writeLock.lock();
try {
-
logger.info("Disconnecting node due to {}", explanation);
- // mark node as not connected
controller.setConnectionStatus(new NodeConnectionStatus(nodeId,
DisconnectionCode.UNKNOWN, explanation));
-
- // turn off primary flag
controller.setPrimary(false);
-
- // stop heartbeating
controller.stopHeartbeating();
-
- // set node to not clustered
controller.setClustered(false, null);
clusterCoordinator.setConnected(false);
logger.info("Node disconnected due to {}", explanation);
-
} finally {
writeLock.unlock();
}
@@ -912,6 +923,8 @@ public class StandardFlowService implements FlowService,
ProtocolHandler {
private void loadFromConnectionResponse(final ConnectionResponse response)
throws ConnectionException {
writeLock.lock();
try {
+ connectionGeneration.incrementAndGet();
+
if (response.getNodeConnectionStatuses() != null) {
clusterCoordinator.resetNodeStatuses(response.getNodeConnectionStatuses().stream()
.collect(Collectors.toMap(NodeConnectionStatus::getNodeIdentifier, status ->
status)));
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestStandardFlowService.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestStandardFlowService.java
new file mode 100644
index 00000000000..1e1a8198e34
--- /dev/null
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/TestStandardFlowService.java
@@ -0,0 +1,135 @@
+/*
+ * 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.controller;
+
+import org.apache.nifi.asset.AssetSynchronizer;
+import org.apache.nifi.authorization.Authorizer;
+import org.apache.nifi.cluster.coordination.ClusterCoordinator;
+import org.apache.nifi.cluster.coordination.node.NodeConnectionStatus;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.cluster.protocol.impl.NodeProtocolSenderListener;
+import org.apache.nifi.cluster.protocol.message.DisconnectMessage;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateManagerProvider;
+import org.apache.nifi.nar.ExtensionManager;
+import org.apache.nifi.nar.NarManager;
+import org.apache.nifi.state.MockStateMap;
+import org.apache.nifi.util.NiFiProperties;
+import org.apache.nifi.web.revision.RevisionManager;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class TestStandardFlowService {
+
+ private FlowController controller;
+ private ClusterCoordinator clusterCoordinator;
+ private StandardFlowService flowService;
+
+ @TempDir
+ private Path tempDir;
+
+ @BeforeEach
+ public void setup() throws IOException {
+ controller = mock(FlowController.class);
+ clusterCoordinator = mock(ClusterCoordinator.class);
+ final NodeProtocolSenderListener senderListener =
mock(NodeProtocolSenderListener.class);
+ final RevisionManager revisionManager = mock(RevisionManager.class);
+ final NarManager narManager = mock(NarManager.class);
+ final AssetSynchronizer parameterContextAssetSynchronizer =
mock(AssetSynchronizer.class);
+ final AssetSynchronizer connectorAssetSynchronizer =
mock(AssetSynchronizer.class);
+ final Authorizer authorizer = mock(Authorizer.class);
+
+ final StateManagerProvider stateManagerProvider =
mock(StateManagerProvider.class);
+ final StateManager stateManager = mock(StateManager.class);
+ when(stateManager.getState(any(Scope.class))).thenReturn(new
MockStateMap(Collections.emptyMap(), 1));
+
when(stateManagerProvider.getStateManager(anyString())).thenReturn(stateManager);
+
when(controller.getStateManagerProvider()).thenReturn(stateManagerProvider);
+
when(controller.getExtensionManager()).thenReturn(mock(ExtensionManager.class));
+
+ final Path flowConfigFile = tempDir.resolve("flow.json.gz");
+ final NiFiProperties nifiProperties =
NiFiProperties.createBasicNiFiProperties(null, Map.of(
+ NiFiProperties.FLOW_CONFIGURATION_FILE,
flowConfigFile.toString(),
+ NiFiProperties.FLOW_CONTROLLER_GRACEFUL_SHUTDOWN_PERIOD, "10
secs",
+ NiFiProperties.WEB_HTTPS_HOST, "localhost",
+ NiFiProperties.WEB_HTTPS_PORT, "8443",
+ NiFiProperties.CLUSTER_NODE_ADDRESS, "localhost",
+ NiFiProperties.CLUSTER_NODE_PROTOCOL_PORT, "9090",
+ NiFiProperties.LOAD_BALANCE_HOST, "localhost",
+ NiFiProperties.LOAD_BALANCE_PORT, "6342",
+ NiFiProperties.FLOW_CONFIGURATION_ARCHIVE_ENABLED, "false"
+ ));
+
+ flowService = StandardFlowService.createClusteredInstance(controller,
nifiProperties, senderListener,
+ clusterCoordinator, revisionManager, narManager,
parameterContextAssetSynchronizer,
+ connectorAssetSynchronizer, authorizer);
+ }
+
+ @Test
+ public void testDisconnectionRequestWithMatchingGeneration() {
+ assertEquals(0, flowService.getConnectionGeneration());
+
+ final DisconnectMessage disconnectMessage =
createDisconnectMessage("Test disconnect");
+ flowService.handleDisconnectionRequest(disconnectMessage, 0);
+
+
verify(controller).setConnectionStatus(any(NodeConnectionStatus.class));
+ verify(controller).setClustered(false, null);
+ verify(clusterCoordinator).setConnected(false);
+ }
+
+ @Test
+ public void testDisconnectionRequestWithStaleGeneration() {
+ assertEquals(0, flowService.getConnectionGeneration());
+
+ final DisconnectMessage disconnectMessage =
createDisconnectMessage("Test disconnect");
+ flowService.handleDisconnectionRequest(disconnectMessage, 0);
+
+ verify(controller).setClustered(false, null);
+
+ reset(controller, clusterCoordinator);
+
+ final DisconnectMessage staleDisconnectMessage =
createDisconnectMessage("Stale disconnect from before reconnect");
+ flowService.handleDisconnectionRequest(staleDisconnectMessage, 5);
+
+ verify(controller, never()).setClustered(false, null);
+ verify(clusterCoordinator, never()).setConnected(false);
+ }
+
+ private DisconnectMessage createDisconnectMessage(final String
explanation) {
+ final NodeIdentifier nodeIdentifier = new NodeIdentifier("node-1",
"localhost", 8443,
+ "localhost", 9090, "localhost", 6342, 10443, false);
+ final DisconnectMessage message = new DisconnectMessage();
+ message.setNodeId(nodeIdentifier);
+ message.setExplanation(explanation);
+ return message;
+ }
+}
