You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2018/10/11 13:52:54 UTC
[3/6] nifi git commit: NIFI-5585 Added capability to offload a node
that is disconnected from the cluster. Updated NodeClusterCoordinator to
allow idempotent requests to offload a cluster Added capability to
connect/delete/disconnect/offload a node from
NIFI-5585 Added capability to offload a node that is disconnected from the cluster.
Updated NodeClusterCoordinator to allow idempotent requests to offload a cluster
Added capability to connect/delete/disconnect/offload a node from the cluster to the Toolkit CLI
Added capability to get the status of nodes from the cluster to the Toolkit CLI
Upgraded FontAwesome to 4.7.0 (from 4.6.1)
Added icon "fa-upload" for offloading nodes in the cluster table UI
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/04d8da8f
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/04d8da8f
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/04d8da8f
Branch: refs/heads/master
Commit: 04d8da8f46c26bd829c4411ea92692589d93278a
Parents: 83ca676
Author: Jeff Storck <jt...@gmail.com>
Authored: Tue Sep 18 17:09:13 2018 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Thu Oct 11 09:23:00 2018 -0400
----------------------------------------------------------------------
.../nifi/controller/queue/FlowFileQueue.java | 2 +
.../coordination/ClusterCoordinator.java | 18 ++++
.../ClusterTopologyEventListener.java | 2 +
.../coordination/node/NodeConnectionState.java | 10 +++
.../coordination/node/NodeConnectionStatus.java | 45 ++++++----
.../cluster/coordination/node/OffloadCode.java | 40 +++++++++
.../ClusterCoordinationProtocolSender.java | 9 ++
...usterCoordinationProtocolSenderListener.java | 6 ++
.../protocol/impl/SocketProtocolListener.java | 3 +
...andardClusterCoordinationProtocolSender.java | 26 ++++++
.../message/AdaptedNodeConnectionStatus.java | 20 +++--
.../message/NodeConnectionStatusAdapter.java | 6 +-
.../protocol/jaxb/message/ObjectFactory.java | 5 ++
.../protocol/message/OffloadMessage.java | 53 +++++++++++
.../protocol/message/ProtocolMessage.java | 1 +
.../heartbeat/AbstractHeartbeatMonitor.java | 8 +-
.../ThreadPoolRequestReplicator.java | 18 ++++
.../node/NodeClusterCoordinator.java | 94 ++++++++++++++++++--
.../exception/IllegalNodeOffloadException.java | 38 ++++++++
.../OffloadedNodeMutableRequestException.java | 39 ++++++++
.../heartbeat/TestAbstractHeartbeatMonitor.java | 11 +++
.../node/TestNodeClusterCoordinator.java | 4 +-
.../nifi/controller/StandardFlowService.java | 71 +++++++++++++--
.../controller/queue/StandardFlowFileQueue.java | 4 +
.../SocketLoadBalancedFlowFileQueue.java | 34 +++++++
.../partition/NonLocalPartitionPartitioner.java | 58 ++++++++++++
.../TestWriteAheadFlowFileRepository.java | 4 +
.../nifi/web/StandardNiFiServiceFacade.java | 6 +-
.../IllegalNodeOffloadExceptionMapper.java | 46 ++++++++++
.../nifi-web-ui/src/main/frontend/package.json | 2 +-
.../webapp/js/nf/cluster/nf-cluster-table.js | 60 ++++++++++++-
.../cli/impl/client/nifi/ControllerClient.java | 14 +++
.../nifi/impl/JerseyControllerClient.java | 87 ++++++++++++++++++
.../toolkit/cli/impl/command/CommandOption.java | 3 +
.../cli/impl/command/nifi/NiFiCommandGroup.java | 12 +++
.../impl/command/nifi/nodes/ConnectNode.java | 67 ++++++++++++++
.../cli/impl/command/nifi/nodes/DeleteNode.java | 58 ++++++++++++
.../impl/command/nifi/nodes/DisconnectNode.java | 67 ++++++++++++++
.../cli/impl/command/nifi/nodes/GetNode.java | 59 ++++++++++++
.../cli/impl/command/nifi/nodes/GetNodes.java | 52 +++++++++++
.../impl/command/nifi/nodes/OffloadNode.java | 67 ++++++++++++++
.../toolkit/cli/impl/result/NodeResult.java | 48 ++++++++++
.../toolkit/cli/impl/result/NodesResult.java | 66 ++++++++++++++
43 files changed, 1299 insertions(+), 44 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-framework-api/src/main/java/org/apache/nifi/controller/queue/FlowFileQueue.java
----------------------------------------------------------------------
diff --git a/nifi-framework-api/src/main/java/org/apache/nifi/controller/queue/FlowFileQueue.java b/nifi-framework-api/src/main/java/org/apache/nifi/controller/queue/FlowFileQueue.java
index 2c7f55b..7cd0e30 100644
--- a/nifi-framework-api/src/main/java/org/apache/nifi/controller/queue/FlowFileQueue.java
+++ b/nifi-framework-api/src/main/java/org/apache/nifi/controller/queue/FlowFileQueue.java
@@ -267,6 +267,8 @@ public interface FlowFileQueue {
void setLoadBalanceStrategy(LoadBalanceStrategy strategy, String partitioningAttribute);
+ void offloadQueue();
+
LoadBalanceStrategy getLoadBalanceStrategy();
void setLoadBalanceCompression(LoadBalanceCompression compression);
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/ClusterCoordinator.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/ClusterCoordinator.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/ClusterCoordinator.java
index 11786c2..2ad0e70 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/ClusterCoordinator.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/ClusterCoordinator.java
@@ -17,6 +17,7 @@
package org.apache.nifi.cluster.coordination;
+import org.apache.nifi.cluster.coordination.node.OffloadCode;
import org.apache.nifi.cluster.coordination.node.DisconnectionCode;
import org.apache.nifi.cluster.coordination.node.NodeConnectionState;
import org.apache.nifi.cluster.coordination.node.NodeConnectionStatus;
@@ -62,6 +63,23 @@ public interface ClusterCoordinator {
void finishNodeConnection(NodeIdentifier nodeId);
/**
+ * Indicates that the node has finished being offloaded
+ *
+ * @param nodeId the identifier of the node
+ */
+ void finishNodeOffload(NodeIdentifier nodeId);
+
+ /**
+ * Sends a request to the node to be offloaded.
+ * The node will be marked as offloading immediately.
+ *
+ * @param nodeId the identifier of the node
+ * @param offloadCode the code that represents why this node is being asked to be offloaded
+ * @param explanation an explanation as to why the node is being asked to be offloaded
+ */
+ void requestNodeOffload(NodeIdentifier nodeId, OffloadCode offloadCode, String explanation);
+
+ /**
* Sends a request to the node to disconnect from the cluster.
* The node will be marked as disconnected immediately.
*
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/ClusterTopologyEventListener.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/ClusterTopologyEventListener.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/ClusterTopologyEventListener.java
index 54cc4de..ad9be3d 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/ClusterTopologyEventListener.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/ClusterTopologyEventListener.java
@@ -23,6 +23,8 @@ public interface ClusterTopologyEventListener {
void onNodeAdded(NodeIdentifier nodeId);
+ void onNodeOffloaded(NodeIdentifier nodeId);
+
void onNodeRemoved(NodeIdentifier nodeId);
void onLocalNodeIdentifierSet(NodeIdentifier localNodeId);
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/NodeConnectionState.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/NodeConnectionState.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/NodeConnectionState.java
index 8d5824f..d79552c 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/NodeConnectionState.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/NodeConnectionState.java
@@ -37,12 +37,22 @@ public enum NodeConnectionState {
CONNECTED,
/**
+ * A node that is in the process of offloading its flow files from the node.
+ */
+ OFFLOADING,
+
+ /**
* A node that is in the process of disconnecting from the cluster.
* A DISCONNECTING node will always transition to DISCONNECTED.
*/
DISCONNECTING,
/**
+ * A node that has offloaded its flow files from the node.
+ */
+ OFFLOADED,
+
+ /**
* A node that is not connected to the cluster.
* A DISCONNECTED node can transition to CONNECTING.
*/
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/NodeConnectionStatus.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/NodeConnectionStatus.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/NodeConnectionStatus.java
index 34bd127..7d8a940 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/NodeConnectionStatus.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/NodeConnectionStatus.java
@@ -35,47 +35,53 @@ public class NodeConnectionStatus {
private final long updateId;
private final NodeIdentifier nodeId;
private final NodeConnectionState state;
+ private final OffloadCode offloadCode;
private final DisconnectionCode disconnectCode;
- private final String disconnectReason;
+ private final String reason;
private final Long connectionRequestTime;
public NodeConnectionStatus(final NodeIdentifier nodeId, final NodeConnectionState state) {
- this(nodeId, state, null, null, null);
+ this(nodeId, state, null, null, null, null);
}
public NodeConnectionStatus(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode) {
- this(nodeId, NodeConnectionState.DISCONNECTED, disconnectionCode, disconnectionCode.toString(), null);
+ this(nodeId, NodeConnectionState.DISCONNECTED, null, disconnectionCode, disconnectionCode.toString(), null);
+ }
+
+ public NodeConnectionStatus(final NodeIdentifier nodeId, final NodeConnectionState state, final OffloadCode offloadCode, final String offloadExplanation) {
+ this(nodeId, state, offloadCode, null, offloadExplanation, null);
}
public NodeConnectionStatus(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String disconnectionExplanation) {
- this(nodeId, NodeConnectionState.DISCONNECTED, disconnectionCode, disconnectionExplanation, null);
+ this(nodeId, NodeConnectionState.DISCONNECTED, null, disconnectionCode, disconnectionExplanation, null);
}
public NodeConnectionStatus(final NodeIdentifier nodeId, final NodeConnectionState state, final DisconnectionCode disconnectionCode) {
- this(nodeId, state, disconnectionCode, disconnectionCode == null ? null : disconnectionCode.toString(), null);
+ this(nodeId, state, null, disconnectionCode, disconnectionCode == null ? null : disconnectionCode.toString(), null);
}
public NodeConnectionStatus(final NodeConnectionStatus status) {
- this(status.getNodeIdentifier(), status.getState(), status.getDisconnectCode(), status.getDisconnectReason(), status.getConnectionRequestTime());
+ this(status.getNodeIdentifier(), status.getState(), status.getOffloadCode(), status.getDisconnectCode(), status.getReason(), status.getConnectionRequestTime());
}
- public NodeConnectionStatus(final NodeIdentifier nodeId, final NodeConnectionState state, final DisconnectionCode disconnectCode,
- final String disconnectReason, final Long connectionRequestTime) {
- this(idGenerator.getAndIncrement(), nodeId, state, disconnectCode, disconnectReason, connectionRequestTime);
+ public NodeConnectionStatus(final NodeIdentifier nodeId, final NodeConnectionState state, final OffloadCode offloadCode,
+ final DisconnectionCode disconnectCode, final String reason, final Long connectionRequestTime) {
+ this(idGenerator.getAndIncrement(), nodeId, state, offloadCode, disconnectCode, reason, connectionRequestTime);
}
- public NodeConnectionStatus(final long updateId, final NodeIdentifier nodeId, final NodeConnectionState state, final DisconnectionCode disconnectCode,
- final String disconnectReason, final Long connectionRequestTime) {
+ public NodeConnectionStatus(final long updateId, final NodeIdentifier nodeId, final NodeConnectionState state, final OffloadCode offloadCode,
+ final DisconnectionCode disconnectCode, final String reason, final Long connectionRequestTime) {
this.updateId = updateId;
this.nodeId = nodeId;
this.state = state;
+ this.offloadCode = offloadCode;
if (state == NodeConnectionState.DISCONNECTED && disconnectCode == null) {
this.disconnectCode = DisconnectionCode.UNKNOWN;
- this.disconnectReason = this.disconnectCode.toString();
+ this.reason = this.disconnectCode.toString();
} else {
this.disconnectCode = disconnectCode;
- this.disconnectReason = disconnectReason;
+ this.reason = reason;
}
this.connectionRequestTime = (connectionRequestTime == null && state == NodeConnectionState.CONNECTING) ? Long.valueOf(System.currentTimeMillis()) : connectionRequestTime;
@@ -93,12 +99,16 @@ public class NodeConnectionStatus {
return state;
}
+ public OffloadCode getOffloadCode() {
+ return offloadCode;
+ }
+
public DisconnectionCode getDisconnectCode() {
return disconnectCode;
}
- public String getDisconnectReason() {
- return disconnectReason;
+ public String getReason() {
+ return reason;
}
public Long getConnectionRequestTime() {
@@ -110,8 +120,11 @@ public class NodeConnectionStatus {
final StringBuilder sb = new StringBuilder();
final NodeConnectionState state = getState();
sb.append("NodeConnectionStatus[nodeId=").append(nodeId).append(", state=").append(state);
+ if (state == NodeConnectionState.OFFLOADED || state == NodeConnectionState.OFFLOADING) {
+ sb.append(", Offload Code=").append(getOffloadCode()).append(", Offload Reason=").append(getReason());
+ }
if (state == NodeConnectionState.DISCONNECTED || state == NodeConnectionState.DISCONNECTING) {
- sb.append(", Disconnect Code=").append(getDisconnectCode()).append(", Disconnect Reason=").append(getDisconnectReason());
+ sb.append(", Disconnect Code=").append(getDisconnectCode()).append(", Disconnect Reason=").append(getReason());
}
sb.append(", updateId=").append(getUpdateIdentifier());
sb.append("]");
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/OffloadCode.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/OffloadCode.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/OffloadCode.java
new file mode 100644
index 0000000..fb4d30b
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/coordination/node/OffloadCode.java
@@ -0,0 +1,40 @@
+/*
+ * 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.cluster.coordination.node;
+
+/**
+ * An enumeration of the reasons that a node may be offloaded
+ */
+public enum OffloadCode {
+
+ /**
+ * A user explicitly offloaded the node
+ */
+ OFFLOADED("Node Offloaded");
+
+ private final String description;
+
+ OffloadCode(final String description) {
+ this.description = description;
+ }
+
+ @Override
+ public String toString() {
+ return description;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/ClusterCoordinationProtocolSender.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/ClusterCoordinationProtocolSender.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/ClusterCoordinationProtocolSender.java
index 986231e..b5485cc 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/ClusterCoordinationProtocolSender.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/ClusterCoordinationProtocolSender.java
@@ -19,6 +19,7 @@ package org.apache.nifi.cluster.protocol;
import java.util.Set;
import org.apache.nifi.cluster.coordination.node.NodeConnectionStatus;
+import org.apache.nifi.cluster.protocol.message.OffloadMessage;
import org.apache.nifi.cluster.protocol.message.DisconnectMessage;
import org.apache.nifi.cluster.protocol.message.NodeStatusChangeMessage;
import org.apache.nifi.cluster.protocol.message.ReconnectionRequestMessage;
@@ -41,6 +42,14 @@ public interface ClusterCoordinationProtocolSender {
ReconnectionResponseMessage requestReconnection(ReconnectionRequestMessage msg) throws ProtocolException;
/**
+ * Sends an "offload request" message to a node.
+ *
+ * @param msg a message
+ * @throws ProtocolException if communication failed
+ */
+ void offload(OffloadMessage msg) throws ProtocolException;
+
+ /**
* Sends a "disconnection request" message to a node.
*
* @param msg a message
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/ClusterCoordinationProtocolSenderListener.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/ClusterCoordinationProtocolSenderListener.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/ClusterCoordinationProtocolSenderListener.java
index ae3a0e5..74cc6b4 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/ClusterCoordinationProtocolSenderListener.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/ClusterCoordinationProtocolSenderListener.java
@@ -26,6 +26,7 @@ import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.cluster.protocol.ProtocolException;
import org.apache.nifi.cluster.protocol.ProtocolHandler;
import org.apache.nifi.cluster.protocol.ProtocolListener;
+import org.apache.nifi.cluster.protocol.message.OffloadMessage;
import org.apache.nifi.cluster.protocol.message.DisconnectMessage;
import org.apache.nifi.cluster.protocol.message.NodeStatusChangeMessage;
import org.apache.nifi.cluster.protocol.message.ReconnectionRequestMessage;
@@ -101,6 +102,11 @@ public class ClusterCoordinationProtocolSenderListener implements ClusterCoordin
}
@Override
+ public void offload(OffloadMessage msg) throws ProtocolException {
+ sender.offload(msg);
+ }
+
+ @Override
public void disconnect(DisconnectMessage msg) throws ProtocolException {
sender.disconnect(msg);
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListener.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListener.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListener.java
index 9eaffd3..c588a68 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListener.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListener.java
@@ -24,6 +24,7 @@ import org.apache.nifi.cluster.protocol.ProtocolListener;
import org.apache.nifi.cluster.protocol.ProtocolMessageMarshaller;
import org.apache.nifi.cluster.protocol.ProtocolMessageUnmarshaller;
import org.apache.nifi.cluster.protocol.message.ConnectionRequestMessage;
+import org.apache.nifi.cluster.protocol.message.OffloadMessage;
import org.apache.nifi.cluster.protocol.message.DisconnectMessage;
import org.apache.nifi.cluster.protocol.message.FlowRequestMessage;
import org.apache.nifi.cluster.protocol.message.HeartbeatMessage;
@@ -210,6 +211,8 @@ public class SocketProtocolListener extends SocketListener implements ProtocolLi
return ((ConnectionRequestMessage) message).getConnectionRequest().getProposedNodeIdentifier();
case HEARTBEAT:
return ((HeartbeatMessage) message).getHeartbeat().getNodeIdentifier();
+ case OFFLOAD_REQUEST:
+ return ((OffloadMessage) message).getNodeId();
case DISCONNECTION_REQUEST:
return ((DisconnectMessage) message).getNodeId();
case FLOW_REQUEST:
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/StandardClusterCoordinationProtocolSender.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/StandardClusterCoordinationProtocolSender.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/StandardClusterCoordinationProtocolSender.java
index 167ddec..b21068f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/StandardClusterCoordinationProtocolSender.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/StandardClusterCoordinationProtocolSender.java
@@ -36,6 +36,7 @@ import org.apache.nifi.cluster.protocol.ProtocolContext;
import org.apache.nifi.cluster.protocol.ProtocolException;
import org.apache.nifi.cluster.protocol.ProtocolMessageMarshaller;
import org.apache.nifi.cluster.protocol.ProtocolMessageUnmarshaller;
+import org.apache.nifi.cluster.protocol.message.OffloadMessage;
import org.apache.nifi.cluster.protocol.message.DisconnectMessage;
import org.apache.nifi.cluster.protocol.message.NodeConnectionStatusRequestMessage;
import org.apache.nifi.cluster.protocol.message.NodeConnectionStatusResponseMessage;
@@ -129,6 +130,31 @@ public class StandardClusterCoordinationProtocolSender implements ClusterCoordin
}
/**
+ * Requests a node to be offloaded. The configured value for
+ * handshake timeout is applied to the socket before making the request.
+ *
+ * @param msg a message
+ * @throws ProtocolException if the message failed to be sent
+ */
+ @Override
+ public void offload(final OffloadMessage msg) throws ProtocolException {
+ Socket socket = null;
+ try {
+ socket = createSocket(msg.getNodeId(), true);
+
+ // marshal message to output stream
+ try {
+ final ProtocolMessageMarshaller<ProtocolMessage> marshaller = protocolContext.createMarshaller();
+ marshaller.marshal(msg, socket.getOutputStream());
+ } catch (final IOException ioe) {
+ throw new ProtocolException("Failed marshalling '" + msg.getType() + "' protocol message due to: " + ioe, ioe);
+ }
+ } finally {
+ SocketUtils.closeQuietly(socket);
+ }
+ }
+
+ /**
* Requests a node to disconnect from the cluster. The configured value for
* handshake timeout is applied to the socket before making the request.
*
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/AdaptedNodeConnectionStatus.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/AdaptedNodeConnectionStatus.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/AdaptedNodeConnectionStatus.java
index c8c4acf..5eae83e 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/AdaptedNodeConnectionStatus.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/AdaptedNodeConnectionStatus.java
@@ -17,6 +17,7 @@
package org.apache.nifi.cluster.protocol.jaxb.message;
+import org.apache.nifi.cluster.coordination.node.OffloadCode;
import org.apache.nifi.cluster.coordination.node.DisconnectionCode;
import org.apache.nifi.cluster.coordination.node.NodeConnectionState;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
@@ -25,8 +26,9 @@ public class AdaptedNodeConnectionStatus {
private Long updateId;
private NodeIdentifier nodeId;
private NodeConnectionState state;
+ private OffloadCode offloadCode;
private DisconnectionCode disconnectCode;
- private String disconnectReason;
+ private String reason;
private Long connectionRequestTime;
public Long getUpdateId() {
@@ -53,20 +55,28 @@ public class AdaptedNodeConnectionStatus {
this.state = state;
}
+ public OffloadCode getOffloadCode() {
+ return offloadCode;
+ }
+
public DisconnectionCode getDisconnectCode() {
return disconnectCode;
}
+ public void setOffloadCode(OffloadCode offloadCode) {
+ this.offloadCode = offloadCode;
+ }
+
public void setDisconnectCode(DisconnectionCode disconnectCode) {
this.disconnectCode = disconnectCode;
}
- public String getDisconnectReason() {
- return disconnectReason;
+ public String getReason() {
+ return reason;
}
- public void setDisconnectReason(String disconnectReason) {
- this.disconnectReason = disconnectReason;
+ public void setReason(String reason) {
+ this.reason = reason;
}
public Long getConnectionRequestTime() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/NodeConnectionStatusAdapter.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/NodeConnectionStatusAdapter.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/NodeConnectionStatusAdapter.java
index ec209de..47e92e8 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/NodeConnectionStatusAdapter.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/NodeConnectionStatusAdapter.java
@@ -28,8 +28,9 @@ public class NodeConnectionStatusAdapter extends XmlAdapter<AdaptedNodeConnectio
return new NodeConnectionStatus(adapted.getUpdateId(),
adapted.getNodeId(),
adapted.getState(),
+ adapted.getOffloadCode(),
adapted.getDisconnectCode(),
- adapted.getDisconnectReason(),
+ adapted.getReason(),
adapted.getConnectionRequestTime());
}
@@ -40,8 +41,9 @@ public class NodeConnectionStatusAdapter extends XmlAdapter<AdaptedNodeConnectio
adapted.setUpdateId(toAdapt.getUpdateIdentifier());
adapted.setNodeId(toAdapt.getNodeIdentifier());
adapted.setConnectionRequestTime(toAdapt.getConnectionRequestTime());
+ adapted.setOffloadCode(toAdapt.getOffloadCode());
adapted.setDisconnectCode(toAdapt.getDisconnectCode());
- adapted.setDisconnectReason(toAdapt.getDisconnectReason());
+ adapted.setReason(toAdapt.getReason());
adapted.setState(toAdapt.getState());
}
return adapted;
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/ObjectFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/ObjectFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/ObjectFactory.java
index 9a594a4..2f02e5e 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/ObjectFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/ObjectFactory.java
@@ -20,6 +20,7 @@ import javax.xml.bind.annotation.XmlRegistry;
import org.apache.nifi.cluster.protocol.message.ConnectionRequestMessage;
import org.apache.nifi.cluster.protocol.message.ConnectionResponseMessage;
+import org.apache.nifi.cluster.protocol.message.OffloadMessage;
import org.apache.nifi.cluster.protocol.message.DisconnectMessage;
import org.apache.nifi.cluster.protocol.message.FlowRequestMessage;
import org.apache.nifi.cluster.protocol.message.FlowResponseMessage;
@@ -52,6 +53,10 @@ public class ObjectFactory {
return new ReconnectionResponseMessage();
}
+ public OffloadMessage createDecomissionMessage() {
+ return new OffloadMessage();
+ }
+
public DisconnectMessage createDisconnectionMessage() {
return new DisconnectMessage();
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/OffloadMessage.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/OffloadMessage.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/OffloadMessage.java
new file mode 100644
index 0000000..a7acd56
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/OffloadMessage.java
@@ -0,0 +1,53 @@
+/*
+ * 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.cluster.protocol.message;
+
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.cluster.protocol.jaxb.message.NodeIdentifierAdapter;
+
+import javax.xml.bind.annotation.XmlRootElement;
+import javax.xml.bind.annotation.adapters.XmlJavaTypeAdapter;
+
+@XmlRootElement(name = "offloadMessage")
+public class OffloadMessage extends ProtocolMessage {
+
+ private NodeIdentifier nodeId;
+ private String explanation;
+
+ @XmlJavaTypeAdapter(NodeIdentifierAdapter.class)
+ public NodeIdentifier getNodeId() {
+ return nodeId;
+ }
+
+ public void setNodeId(NodeIdentifier nodeId) {
+ this.nodeId = nodeId;
+ }
+
+ public String getExplanation() {
+ return explanation;
+ }
+
+ public void setExplanation(String explanation) {
+ this.explanation = explanation;
+ }
+
+ @Override
+ public MessageType getType() {
+ return MessageType.OFFLOAD_REQUEST;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/ProtocolMessage.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/ProtocolMessage.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/ProtocolMessage.java
index 482f5d6..fe26c7a 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/ProtocolMessage.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/ProtocolMessage.java
@@ -21,6 +21,7 @@ public abstract class ProtocolMessage {
public static enum MessageType {
CONNECTION_REQUEST,
CONNECTION_RESPONSE,
+ OFFLOAD_REQUEST,
DISCONNECTION_REQUEST,
EXCEPTION,
FLOW_REQUEST,
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/heartbeat/AbstractHeartbeatMonitor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/heartbeat/AbstractHeartbeatMonitor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/heartbeat/AbstractHeartbeatMonitor.java
index 4395883..f6d09ab 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/heartbeat/AbstractHeartbeatMonitor.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/heartbeat/AbstractHeartbeatMonitor.java
@@ -228,6 +228,12 @@ public abstract class AbstractHeartbeatMonitor implements HeartbeatMonitor {
return;
}
+ if (NodeConnectionState.OFFLOADED == connectionState) {
+ // Cluster Coordinator believes that node is offloaded, but let the node reconnect
+ clusterCoordinator.reportEvent(nodeId, Severity.INFO, "Received heartbeat from node that is offloaded. " +
+ "Marking as Disconnected and requesting that Node reconnect to cluster");
+ clusterCoordinator.requestNodeConnect(nodeId, null);
+ }
if (NodeConnectionState.DISCONNECTED == connectionState) {
// ignore heartbeats from nodes disconnected by means other than lack of heartbeat, unless it is
// the only node. We allow it if it is the only node because if we have a one-node cluster, then
@@ -249,7 +255,7 @@ public abstract class AbstractHeartbeatMonitor implements HeartbeatMonitor {
default: {
// disconnected nodes should not heartbeat, so we need to issue a disconnection request.
logger.info("Ignoring received heartbeat from disconnected node " + nodeId + ". Issuing disconnection request.");
- clusterCoordinator.requestNodeDisconnect(nodeId, disconnectionCode, connectionStatus.getDisconnectReason());
+ clusterCoordinator.requestNodeDisconnect(nodeId, disconnectionCode, connectionStatus.getReason());
removeHeartbeat(nodeId);
break;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ThreadPoolRequestReplicator.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ThreadPoolRequestReplicator.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ThreadPoolRequestReplicator.java
index 85618de..0b2f1fe 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ThreadPoolRequestReplicator.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/replication/ThreadPoolRequestReplicator.java
@@ -170,6 +170,24 @@ public class ThreadPoolRequestReplicator implements RequestReplicator {
// If the request is mutable, ensure that all nodes are connected.
if (mutable) {
+ final List<NodeIdentifier> offloaded = stateMap.get(NodeConnectionState.OFFLOADED);
+ if (offloaded != null && !offloaded.isEmpty()) {
+ if (offloaded.size() == 1) {
+ throw new OffloadedNodeMutableRequestException("Node " + offloaded.iterator().next() + " is currently offloaded");
+ } else {
+ throw new OffloadedNodeMutableRequestException(offloaded.size() + " Nodes are currently offloaded");
+ }
+ }
+
+ final List<NodeIdentifier> offloading = stateMap.get(NodeConnectionState.OFFLOADING);
+ if (offloading != null && !offloading.isEmpty()) {
+ if (offloading.size() == 1) {
+ throw new OffloadedNodeMutableRequestException("Node " + offloading.iterator().next() + " is currently offloading");
+ } else {
+ throw new OffloadedNodeMutableRequestException(offloading.size() + " Nodes are currently offloading");
+ }
+ }
+
final List<NodeIdentifier> disconnected = stateMap.get(NodeConnectionState.DISCONNECTED);
if (disconnected != null && !disconnected.isEmpty()) {
if (disconnected.size() == 1) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/NodeClusterCoordinator.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/NodeClusterCoordinator.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/NodeClusterCoordinator.java
index b24475e..e165041 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/NodeClusterCoordinator.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/NodeClusterCoordinator.java
@@ -34,6 +34,7 @@ import org.apache.nifi.cluster.event.NodeEvent;
import org.apache.nifi.cluster.exception.NoClusterCoordinatorException;
import org.apache.nifi.cluster.firewall.ClusterNodeFirewall;
import org.apache.nifi.cluster.manager.NodeResponse;
+import org.apache.nifi.cluster.manager.exception.IllegalNodeOffloadException;
import org.apache.nifi.cluster.manager.exception.IllegalNodeDisconnectionException;
import org.apache.nifi.cluster.protocol.ComponentRevision;
import org.apache.nifi.cluster.protocol.ConnectionRequest;
@@ -49,6 +50,7 @@ import org.apache.nifi.cluster.protocol.message.ClusterWorkloadRequestMessage;
import org.apache.nifi.cluster.protocol.message.ClusterWorkloadResponseMessage;
import org.apache.nifi.cluster.protocol.message.ConnectionRequestMessage;
import org.apache.nifi.cluster.protocol.message.ConnectionResponseMessage;
+import org.apache.nifi.cluster.protocol.message.OffloadMessage;
import org.apache.nifi.cluster.protocol.message.DisconnectMessage;
import org.apache.nifi.cluster.protocol.message.NodeConnectionStatusResponseMessage;
import org.apache.nifi.cluster.protocol.message.NodeStatusChangeMessage;
@@ -431,7 +433,7 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl
reportEvent(nodeId, Severity.INFO, "Requesting that node connect to cluster on behalf of " + userDn);
}
- updateNodeStatus(new NodeConnectionStatus(nodeId, NodeConnectionState.CONNECTING, null, null, System.currentTimeMillis()));
+ updateNodeStatus(new NodeConnectionStatus(nodeId, NodeConnectionState.CONNECTING, null, null, null, System.currentTimeMillis()));
// create the request
final ReconnectionRequestMessage request = new ReconnectionRequestMessage();
@@ -470,6 +472,50 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl
}
@Override
+ public void finishNodeOffload(final NodeIdentifier nodeId) {
+ final NodeConnectionState state = getConnectionState(nodeId);
+ if (state == null) {
+ logger.warn("Attempted to finish node offload for {} but node is not known.", nodeId);
+ return;
+ }
+
+ if (state != NodeConnectionState.OFFLOADING) {
+ logger.warn("Attempted to finish node offload for {} but node is not in a offload state, it is currently {}.", nodeId, state);
+ return;
+ }
+
+ logger.info("{} is now offloaded", nodeId);
+ updateNodeStatus(new NodeConnectionStatus(nodeId, NodeConnectionState.OFFLOADED));
+ }
+
+ @Override
+ public void requestNodeOffload(final NodeIdentifier nodeId, final OffloadCode offloadCode, final String explanation) {
+ final Set<NodeIdentifier> offloadNodeIds = getNodeIdentifiers(NodeConnectionState.OFFLOADING, NodeConnectionState.OFFLOADED);
+ if (offloadNodeIds.contains(nodeId)) {
+ logger.debug("Attempted to offload node but the node is already offloading or offloaded");
+ // no need to do anything here, the node is currently offloading or already offloaded
+ return;
+ }
+
+ final Set<NodeIdentifier> disconnectedNodeIds = getNodeIdentifiers(NodeConnectionState.DISCONNECTED);
+ if (!disconnectedNodeIds.contains(nodeId)) {
+ throw new IllegalNodeOffloadException("Cannot offload node " + nodeId + " because it is not currently disconnected");
+ }
+
+ logger.info("Requesting that {} is offloaded due to {}", nodeId, explanation == null ? offloadCode : explanation);
+
+ updateNodeStatus(new NodeConnectionStatus(nodeId, NodeConnectionState.OFFLOADING, offloadCode, explanation));
+
+ final OffloadMessage request = new OffloadMessage();
+ request.setNodeId(nodeId);
+ request.setExplanation(explanation);
+
+ addNodeEvent(nodeId, "Offload requested due to " + explanation);
+ onNodeOffloaded(nodeId);
+ offloadAsynchronously(request, 10, 5);
+ }
+
+ @Override
public void requestNodeDisconnect(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String explanation) {
final Set<NodeIdentifier> connectedNodeIds = getNodeIdentifiers(NodeConnectionState.CONNECTED);
if (connectedNodeIds.size() == 1 && connectedNodeIds.contains(nodeId)) {
@@ -526,17 +572,19 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl
storeState();
}
+ private void onNodeOffloaded(final NodeIdentifier nodeId) {
+ eventListeners.forEach(listener -> listener.onNodeOffloaded(nodeId));
+ }
+
private void onNodeRemoved(final NodeIdentifier nodeId) {
- eventListeners.stream().forEach(listener -> listener.onNodeRemoved(nodeId));
+ eventListeners.forEach(listener -> listener.onNodeRemoved(nodeId));
}
private void onNodeAdded(final NodeIdentifier nodeId, final boolean storeState) {
if (storeState) {
storeState();
}
-
-
- eventListeners.stream().forEach(listener -> listener.onNodeAdded(nodeId));
+ eventListeners.forEach(listener -> listener.onNodeAdded(nodeId));
}
@Override
@@ -821,7 +869,7 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl
// Otherwise, get the active coordinator (or wait for one to become active) and then notify the coordinator.
final Set<NodeIdentifier> nodesToNotify;
if (notifyAllNodes) {
- nodesToNotify = getNodeIdentifiers(NodeConnectionState.CONNECTED, NodeConnectionState.CONNECTING);
+ nodesToNotify = getNodeIdentifiers();
// Do not notify ourselves because we already know about the status update.
nodesToNotify.remove(getLocalNodeIdentifier());
@@ -841,6 +889,34 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl
senderListener.notifyNodeStatusChange(nodesToNotify, message);
}
+ private void offloadAsynchronously(final OffloadMessage request, final int attempts, final int retrySeconds) {
+ final Thread offloadThread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ final NodeIdentifier nodeId = request.getNodeId();
+
+ for (int i = 0; i < attempts; i++) {
+ try {
+ senderListener.offload(request);
+ reportEvent(nodeId, Severity.INFO, "Node was offloaded due to " + request.getExplanation());
+ return;
+ } catch (final Exception e) {
+ logger.error("Failed to notify {} that it has been offloaded due to {}", request.getNodeId(), request.getExplanation(), e);
+
+ try {
+ Thread.sleep(retrySeconds * 1000L);
+ } catch (final InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+ }
+ }
+ }, "Offload " + request.getNodeId());
+
+ offloadThread.start();
+ }
+
private void disconnectAsynchronously(final DisconnectMessage request, final int attempts, final int retrySeconds) {
final Thread disconnectThread = new Thread(new Runnable() {
@Override
@@ -961,8 +1037,8 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl
if (oldStatus == null || status.getState() != oldStatus.getState()) {
sb.append("Node Status changed from ").append(oldStatus == null ? "[Unknown Node]" : oldStatus.getState().toString()).append(" to ").append(status.getState().toString());
- if (status.getDisconnectReason() != null) {
- sb.append(" due to ").append(status.getDisconnectReason());
+ if (status.getReason() != null) {
+ sb.append(" due to ").append(status.getReason());
} else if (status.getDisconnectCode() != null) {
sb.append(" due to ").append(status.getDisconnectCode().toString());
}
@@ -1118,7 +1194,7 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl
addNodeEvent(resolvedNodeIdentifier, "Connection requested from existing node. Setting status to connecting.");
}
- status = new NodeConnectionStatus(resolvedNodeIdentifier, NodeConnectionState.CONNECTING, null, null, System.currentTimeMillis());
+ status = new NodeConnectionStatus(resolvedNodeIdentifier, NodeConnectionState.CONNECTING, null, null, null, System.currentTimeMillis());
updateNodeStatus(status);
final ConnectionResponse response = new ConnectionResponse(resolvedNodeIdentifier, clusterDataFlow, instanceId, getConnectionStatuses(),
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/IllegalNodeOffloadException.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/IllegalNodeOffloadException.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/IllegalNodeOffloadException.java
new file mode 100644
index 0000000..f1bc669
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/IllegalNodeOffloadException.java
@@ -0,0 +1,38 @@
+/*
+ * 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.cluster.manager.exception;
+
+/**
+ * Represents the exceptional case when an offload request is issued to a node that cannot be offloaded (e.g., not currently disconnected).
+ */
+public class IllegalNodeOffloadException extends IllegalClusterStateException {
+
+ public IllegalNodeOffloadException() {
+ }
+
+ public IllegalNodeOffloadException(String msg) {
+ super(msg);
+ }
+
+ public IllegalNodeOffloadException(Throwable cause) {
+ super(cause);
+ }
+
+ public IllegalNodeOffloadException(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/OffloadedNodeMutableRequestException.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/OffloadedNodeMutableRequestException.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/OffloadedNodeMutableRequestException.java
new file mode 100644
index 0000000..3663349
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/exception/OffloadedNodeMutableRequestException.java
@@ -0,0 +1,39 @@
+/*
+ * 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.cluster.manager.exception;
+
+/**
+ * Represents the exceptional case when a HTTP request that may change a node's dataflow is to be replicated while one or more nodes are offloaded.
+ *
+ */
+public class OffloadedNodeMutableRequestException extends MutableRequestException {
+
+ public OffloadedNodeMutableRequestException() {
+ }
+
+ public OffloadedNodeMutableRequestException(String msg) {
+ super(msg);
+ }
+
+ public OffloadedNodeMutableRequestException(Throwable cause) {
+ super(cause);
+ }
+
+ public OffloadedNodeMutableRequestException(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/heartbeat/TestAbstractHeartbeatMonitor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/heartbeat/TestAbstractHeartbeatMonitor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/heartbeat/TestAbstractHeartbeatMonitor.java
index 6ea019d..4aeff7b 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/heartbeat/TestAbstractHeartbeatMonitor.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/heartbeat/TestAbstractHeartbeatMonitor.java
@@ -20,6 +20,7 @@ package org.apache.nifi.cluster.coordination.heartbeat;
import org.apache.nifi.cluster.ReportedEvent;
import org.apache.nifi.cluster.coordination.ClusterCoordinator;
import org.apache.nifi.cluster.coordination.ClusterTopologyEventListener;
+import org.apache.nifi.cluster.coordination.node.OffloadCode;
import org.apache.nifi.cluster.coordination.node.DisconnectionCode;
import org.apache.nifi.cluster.coordination.node.NodeConnectionState;
import org.apache.nifi.cluster.coordination.node.NodeConnectionStatus;
@@ -245,6 +246,16 @@ public class TestAbstractHeartbeatMonitor {
}
@Override
+ public synchronized void finishNodeOffload(NodeIdentifier nodeId) {
+ statuses.put(nodeId, new NodeConnectionStatus(nodeId, NodeConnectionState.OFFLOADED));
+ }
+
+ @Override
+ public synchronized void requestNodeOffload(NodeIdentifier nodeId, OffloadCode offloadCode, String explanation) {
+ statuses.put(nodeId, new NodeConnectionStatus(nodeId, NodeConnectionState.OFFLOADED));
+ }
+
+ @Override
public synchronized void requestNodeDisconnect(NodeIdentifier nodeId, DisconnectionCode disconnectionCode, String explanation) {
statuses.put(nodeId, new NodeConnectionStatus(nodeId, NodeConnectionState.DISCONNECTED));
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/node/TestNodeClusterCoordinator.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/node/TestNodeClusterCoordinator.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/node/TestNodeClusterCoordinator.java
index fb06a15..5ce2985 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/node/TestNodeClusterCoordinator.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/node/TestNodeClusterCoordinator.java
@@ -280,7 +280,7 @@ public class TestNodeClusterCoordinator {
assertNotNull(statusChange);
assertEquals(createNodeId(1), statusChange.getNodeIdentifier());
assertEquals(DisconnectionCode.NODE_SHUTDOWN, statusChange.getDisconnectCode());
- assertEquals("Unit Test", statusChange.getDisconnectReason());
+ assertEquals("Unit Test", statusChange.getReason());
}
@Test
@@ -407,7 +407,7 @@ public class TestNodeClusterCoordinator {
nodeStatuses.clear();
final NodeConnectionStatus oldStatus = new NodeConnectionStatus(-1L, nodeId1, NodeConnectionState.DISCONNECTED,
- DisconnectionCode.BLOCKED_BY_FIREWALL, null, 0L);
+ null, DisconnectionCode.BLOCKED_BY_FIREWALL, null, 0L);
final NodeStatusChangeMessage msg = new NodeStatusChangeMessage();
msg.setNodeId(nodeId1);
msg.setNodeConnectionStatus(oldStatus);
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
index 7a5c45e..297595f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/StandardFlowService.java
@@ -23,6 +23,7 @@ import org.apache.nifi.authorization.ManagedAuthorizer;
import org.apache.nifi.bundle.Bundle;
import org.apache.nifi.cluster.ConnectionException;
import org.apache.nifi.cluster.coordination.ClusterCoordinator;
+import org.apache.nifi.cluster.coordination.node.OffloadCode;
import org.apache.nifi.cluster.coordination.node.DisconnectionCode;
import org.apache.nifi.cluster.coordination.node.NodeConnectionState;
import org.apache.nifi.cluster.coordination.node.NodeConnectionStatus;
@@ -36,6 +37,7 @@ import org.apache.nifi.cluster.protocol.ProtocolHandler;
import org.apache.nifi.cluster.protocol.StandardDataFlow;
import org.apache.nifi.cluster.protocol.impl.NodeProtocolSenderListener;
import org.apache.nifi.cluster.protocol.message.ConnectionRequestMessage;
+import org.apache.nifi.cluster.protocol.message.OffloadMessage;
import org.apache.nifi.cluster.protocol.message.DisconnectMessage;
import org.apache.nifi.cluster.protocol.message.FlowRequestMessage;
import org.apache.nifi.cluster.protocol.message.FlowResponseMessage;
@@ -44,12 +46,14 @@ import org.apache.nifi.cluster.protocol.message.ReconnectionRequestMessage;
import org.apache.nifi.cluster.protocol.message.ReconnectionResponseMessage;
import org.apache.nifi.components.state.Scope;
import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.controller.queue.FlowFileQueue;
import org.apache.nifi.controller.serialization.FlowSerializationException;
import org.apache.nifi.controller.serialization.FlowSynchronizationException;
import org.apache.nifi.encrypt.StringEncryptor;
import org.apache.nifi.engine.FlowEngine;
import org.apache.nifi.events.BulletinFactory;
import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.lifecycle.LifeCycleStartException;
import org.apache.nifi.logging.LogLevel;
import org.apache.nifi.nar.NarClassLoaders;
@@ -381,6 +385,7 @@ public class StandardFlowService implements FlowService, ProtocolHandler {
public boolean canHandle(final ProtocolMessage msg) {
switch (msg.getType()) {
case RECONNECTION_REQUEST:
+ case OFFLOAD_REQUEST:
case DISCONNECTION_REQUEST:
case FLOW_REQUEST:
return true;
@@ -415,6 +420,22 @@ public class StandardFlowService implements FlowService, ProtocolHandler {
return new ReconnectionResponseMessage();
}
+ case OFFLOAD_REQUEST: {
+ final Thread t = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ handleOffloadRequest((OffloadMessage) request);
+ } catch (InterruptedException e) {
+ throw new ProtocolException("Could not complete offload request", e);
+ }
+ }
+ }, "Offload Flow Files from Node");
+ t.setDaemon(true);
+ t.start();
+
+ return null;
+ }
case DISCONNECTION_REQUEST: {
final Thread t = new Thread(new Runnable() {
@Override
@@ -561,7 +582,7 @@ public class StandardFlowService implements FlowService, ProtocolHandler {
private FlowResponseMessage handleFlowRequest(final FlowRequestMessage request) throws ProtocolException {
readLock.lock();
try {
- logger.info("Received flow request message from manager.");
+ logger.info("Received flow request message from cluster coordinator.");
// create the response
final FlowResponseMessage response = new FlowResponseMessage();
@@ -631,7 +652,7 @@ public class StandardFlowService implements FlowService, ProtocolHandler {
private void handleReconnectionRequest(final ReconnectionRequestMessage request) {
try {
- logger.info("Processing reconnection request from manager.");
+ logger.info("Processing reconnection request from cluster coordinator.");
// reconnect
ConnectionResponse connectionResponse = new ConnectionResponse(getNodeId(), request.getDataFlow(),
@@ -662,8 +683,48 @@ public class StandardFlowService implements FlowService, ProtocolHandler {
}
}
+ private void handleOffloadRequest(final OffloadMessage request) throws InterruptedException {
+ logger.info("Received offload request message from cluster coordinator with explanation: " + request.getExplanation());
+ offload(request.getExplanation());
+ }
+
+ private void offload(final String explanation) throws InterruptedException {
+ writeLock.lock();
+ try {
+
+ logger.info("Offloading node due to " + explanation);
+
+ // mark node as offloading
+ controller.setConnectionStatus(new NodeConnectionStatus(nodeId, NodeConnectionState.OFFLOADING, OffloadCode.OFFLOADED, explanation));
+ // request to stop all processors on node
+ controller.stopAllProcessors();
+ // request to stop all remote process groups
+ controller.getRootGroup().findAllRemoteProcessGroups().forEach(RemoteProcessGroup::stopTransmitting);
+ // terminate all processors
+ controller.getRootGroup().findAllProcessors()
+ // filter stream, only stopped processors can be terminated
+ .stream().filter(pn -> pn.getScheduledState() == ScheduledState.STOPPED)
+ .forEach(pn -> pn.getProcessGroup().terminateProcessor(pn));
+ // offload all queues on node
+ controller.getAllQueues().forEach(FlowFileQueue::offloadQueue);
+ // wait for rebalance of flowfiles on all queues
+ while (controller.getControllerStatus().getQueuedCount() > 0) {
+ logger.debug("Offloading queues on node {}, remaining queued count: {}", getNodeId(), controller.getControllerStatus().getQueuedCount());
+ Thread.sleep(1000);
+ }
+ // finish offload
+ controller.setConnectionStatus(new NodeConnectionStatus(nodeId, NodeConnectionState.OFFLOADED, OffloadCode.OFFLOADED, explanation));
+ clusterCoordinator.finishNodeOffload(getNodeId());
+
+ logger.info("Node offloaded due to " + explanation);
+
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
private void handleDisconnectionRequest(final DisconnectMessage request) {
- logger.info("Received disconnection request message from manager with explanation: " + request.getExplanation());
+ logger.info("Received disconnection request message from cluster coordinator with explanation: " + request.getExplanation());
disconnect(request.getExplanation());
}
@@ -829,11 +890,11 @@ public class StandardFlowService implements FlowService, ProtocolHandler {
}
} else if (response.getRejectionReason() != null) {
logger.warn("Connection request was blocked by cluster coordinator with the explanation: " + response.getRejectionReason());
- // set response to null and treat a firewall blockage the same as getting no response from manager
+ // set response to null and treat a firewall blockage the same as getting no response from cluster coordinator
response = null;
break;
} else {
- // we received a successful connection response from manager
+ // we received a successful connection response from cluster coordinator
break;
}
} catch (final NoClusterCoordinatorException ncce) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/StandardFlowFileQueue.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/StandardFlowFileQueue.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/StandardFlowFileQueue.java
index cab41e8..ee222f4 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/StandardFlowFileQueue.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/StandardFlowFileQueue.java
@@ -75,6 +75,10 @@ public class StandardFlowFileQueue extends AbstractFlowFileQueue implements Flow
}
@Override
+ public void offloadQueue() {
+ }
+
+ @Override
public boolean isActivelyLoadBalancing() {
return false;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
index 193a961..4c9188b 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/SocketLoadBalancedFlowFileQueue.java
@@ -40,6 +40,7 @@ import org.apache.nifi.controller.queue.clustered.partition.FirstNodePartitioner
import org.apache.nifi.controller.queue.clustered.partition.FlowFilePartitioner;
import org.apache.nifi.controller.queue.clustered.partition.LocalPartitionPartitioner;
import org.apache.nifi.controller.queue.clustered.partition.LocalQueuePartition;
+import org.apache.nifi.controller.queue.clustered.partition.NonLocalPartitionPartitioner;
import org.apache.nifi.controller.queue.clustered.partition.QueuePartition;
import org.apache.nifi.controller.queue.clustered.partition.RebalancingPartition;
import org.apache.nifi.controller.queue.clustered.partition.RemoteQueuePartition;
@@ -113,6 +114,7 @@ public class SocketLoadBalancedFlowFileQueue extends AbstractFlowFileQueue imple
private QueuePartition[] queuePartitions;
private FlowFilePartitioner partitioner;
private boolean stopped = true;
+ private boolean offloaded = false;
public SocketLoadBalancedFlowFileQueue(final String identifier, final ConnectionEventListener eventListener, final ProcessScheduler scheduler, final FlowFileRepository flowFileRepo,
@@ -204,6 +206,19 @@ public class SocketLoadBalancedFlowFileQueue extends AbstractFlowFileQueue imple
setFlowFilePartitioner(partitioner);
}
+ @Override
+ public void offloadQueue() {
+ if (clusterCoordinator == null) {
+ // Not clustered, so don't change partitions
+ return;
+ }
+
+ offloaded = true;
+
+ // TODO need to be able to reset the partitioner to the previous partitioner if this node is reconnected to the cluster
+ setFlowFilePartitioner(new NonLocalPartitionPartitioner());
+ }
+
public synchronized void startLoadBalancing() {
logger.debug("{} started. Will begin distributing FlowFiles across the cluster", this);
@@ -551,6 +566,11 @@ public class SocketLoadBalancedFlowFileQueue extends AbstractFlowFileQueue imple
return;
}
+ if (offloaded) {
+ logger.debug("{} Not going to rebalance Queue even though setNodeIdentifiers was called, because the queue has been offloaded", this);
+ return;
+ }
+
logger.debug("{} Stopping the {} queue partitions in order to change node identifiers from {} to {}", this, queuePartitions.length, this.nodeIdentifiers, updatedNodeIdentifiers);
for (final QueuePartition queuePartition : queuePartitions) {
queuePartition.stop();
@@ -969,6 +989,20 @@ public class SocketLoadBalancedFlowFileQueue extends AbstractFlowFileQueue imple
}
@Override
+ public void onNodeOffloaded(final NodeIdentifier nodeId) {
+ partitionWriteLock.lock();
+ try {
+ final Set<NodeIdentifier> updatedNodeIds = new HashSet<>(nodeIdentifiers);
+ updatedNodeIds.remove(nodeId);
+
+ logger.debug("Node Identifier {} offloaded. Node ID's changing from {} to {}", nodeId, nodeIdentifiers, updatedNodeIds);
+ setNodeIdentifiers(updatedNodeIds, false);
+ } finally {
+ partitionWriteLock.unlock();
+ }
+ }
+
+ @Override
public void onNodeRemoved(final NodeIdentifier nodeId) {
partitionWriteLock.lock();
try {
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/partition/NonLocalPartitionPartitioner.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/partition/NonLocalPartitionPartitioner.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/partition/NonLocalPartitionPartitioner.java
new file mode 100644
index 0000000..cffaefd
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/queue/clustered/partition/NonLocalPartitionPartitioner.java
@@ -0,0 +1,58 @@
+/*
+ * 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.queue.clustered.partition;
+
+import org.apache.nifi.controller.repository.FlowFileRecord;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Returns remote partitions when queried for a partition; never returns the {@link LocalQueuePartition}.
+ */
+public class NonLocalPartitionPartitioner implements FlowFilePartitioner {
+ private final AtomicLong counter = new AtomicLong(0L);
+
+ @Override
+ public QueuePartition getPartition(final FlowFileRecord flowFile, final QueuePartition[] partitions, final QueuePartition localPartition) {
+ QueuePartition remotePartition = null;
+ final long startIndex = counter.getAndIncrement();
+ for (int i = 0, numPartitions = partitions.length; i < numPartitions && remotePartition == null; ++i) {
+ int index = (int) ((startIndex + i) % numPartitions);
+ QueuePartition partition = partitions[index];
+ if (!partition.equals(localPartition)) {
+ remotePartition = partition;
+ }
+ }
+
+ if (remotePartition == null) {
+ throw new IllegalStateException("Could not determine a remote partition");
+ }
+
+ return remotePartition;
+ }
+
+ @Override
+ public boolean isRebalanceOnClusterResize() {
+ return false;
+ }
+
+
+ @Override
+ public boolean isRebalanceOnFailure() {
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
index 402ce06..878ad13 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestWriteAheadFlowFileRepository.java
@@ -110,6 +110,10 @@ public class TestWriteAheadFlowFileRepository {
}
@Override
+ public void offloadQueue() {
+ }
+
+ @Override
public boolean isActivelyLoadBalancing() {
return false;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
index b8ac88a..4ef241e 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
@@ -47,6 +47,7 @@ import org.apache.nifi.cluster.coordination.ClusterCoordinator;
import org.apache.nifi.cluster.coordination.heartbeat.HeartbeatMonitor;
import org.apache.nifi.cluster.coordination.heartbeat.NodeHeartbeat;
import org.apache.nifi.cluster.coordination.node.ClusterRoles;
+import org.apache.nifi.cluster.coordination.node.OffloadCode;
import org.apache.nifi.cluster.coordination.node.DisconnectionCode;
import org.apache.nifi.cluster.coordination.node.NodeConnectionState;
import org.apache.nifi.cluster.coordination.node.NodeConnectionStatus;
@@ -1124,6 +1125,9 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
if (NodeConnectionState.CONNECTING.name().equalsIgnoreCase(nodeDTO.getStatus())) {
clusterCoordinator.requestNodeConnect(nodeId, userDn);
+ } else if (NodeConnectionState.OFFLOADING.name().equalsIgnoreCase(nodeDTO.getStatus())) {
+ clusterCoordinator.requestNodeOffload(nodeId, OffloadCode.OFFLOADED,
+ "User " + userDn + " requested that node be offloaded");
} else if (NodeConnectionState.DISCONNECTING.name().equalsIgnoreCase(nodeDTO.getStatus())) {
clusterCoordinator.requestNodeDisconnect(nodeId, DisconnectionCode.USER_DISCONNECTED,
"User " + userDn + " requested that node be disconnected from cluster");
@@ -4702,7 +4706,7 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
}
final NodeConnectionStatus nodeConnectionStatus = clusterCoordinator.getConnectionStatus(nodeIdentifier);
- if (!nodeConnectionStatus.getState().equals(NodeConnectionState.DISCONNECTED)) {
+ if (!nodeConnectionStatus.getState().equals(NodeConnectionState.OFFLOADED) && !nodeConnectionStatus.getState().equals(NodeConnectionState.DISCONNECTED)) {
throw new IllegalNodeDeletionException("Cannot remove Node with ID " + nodeId + " because it is not disconnected, current state = " + nodeConnectionStatus.getState());
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/config/IllegalNodeOffloadExceptionMapper.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/config/IllegalNodeOffloadExceptionMapper.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/config/IllegalNodeOffloadExceptionMapper.java
new file mode 100644
index 0000000..890a8a6
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/config/IllegalNodeOffloadExceptionMapper.java
@@ -0,0 +1,46 @@
+/*
+ * 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.web.api.config;
+
+import org.apache.nifi.cluster.manager.exception.IllegalNodeOffloadException;
+import org.apache.nifi.util.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.ws.rs.core.Response;
+import javax.ws.rs.ext.ExceptionMapper;
+import javax.ws.rs.ext.Provider;
+
+/**
+ * Maps illegal node offload exceptions into client responses.
+ */
+@Provider
+public class IllegalNodeOffloadExceptionMapper implements ExceptionMapper<IllegalNodeOffloadException> {
+
+ private static final Logger logger = LoggerFactory.getLogger(IllegalNodeOffloadExceptionMapper.class);
+
+ @Override
+ public Response toResponse(IllegalNodeOffloadException exception) {
+ logger.info(String.format("%s. Returning %s response.", exception, Response.Status.CONFLICT));
+
+ if (logger.isDebugEnabled()) {
+ logger.debug(StringUtils.EMPTY, exception);
+ }
+ return Response.status(Response.Status.CONFLICT).entity(exception.getMessage()).type("text/plain").build();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/04d8da8f/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-ui/src/main/frontend/package.json
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-ui/src/main/frontend/package.json b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-ui/src/main/frontend/package.json
index eff9e33..0bbd0ed 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-ui/src/main/frontend/package.json
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-ui/src/main/frontend/package.json
@@ -30,7 +30,7 @@
"d3-selection-multi": "1.0.1",
"jquery-minicolors": "2.1.10",
"jquery-ui-dist": "1.12.1",
- "font-awesome": "4.6.1",
+ "font-awesome": "4.7.0",
"jquery": "3.1.1",
"reset.css": "2.0.2",
"jquery-form": "3.50.0",