http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/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 07cefbb..bd4e277 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 @@ -75,6 +75,7 @@ import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.Function; import java.util.stream.Collectors; +import org.apache.nifi.util.NiFiProperties; public class ThreadPoolRequestReplicator implements RequestReplicator { @@ -107,10 +108,11 @@ public class ThreadPoolRequestReplicator implements RequestReplicator { * @param clusterCoordinator the cluster coordinator to use for interacting with node statuses * @param callback a callback that will be called whenever all of the responses have been gathered for a request. May be null. * @param eventReporter an EventReporter that can be used to notify users of interesting events. May be null. + * @param nifiProperties properties */ public ThreadPoolRequestReplicator(final int numThreads, final Client client, final ClusterCoordinator clusterCoordinator, - final RequestCompletionCallback callback, final EventReporter eventReporter) { - this(numThreads, client, clusterCoordinator, "5 sec", "5 sec", callback, eventReporter); + final RequestCompletionCallback callback, final EventReporter eventReporter, final NiFiProperties nifiProperties) { + this(numThreads, client, clusterCoordinator, "5 sec", "5 sec", callback, eventReporter, nifiProperties); } /** @@ -123,9 +125,11 @@ public class ThreadPoolRequestReplicator implements RequestReplicator { * @param readTimeout the read timeout specified in milliseconds * @param callback a callback that will be called whenever all of the responses have been gathered for a request. May be null. * @param eventReporter an EventReporter that can be used to notify users of interesting events. May be null. + * @param nifiProperties properties */ public ThreadPoolRequestReplicator(final int numThreads, final Client client, final ClusterCoordinator clusterCoordinator, - final String connectionTimeout, final String readTimeout, final RequestCompletionCallback callback, final EventReporter eventReporter) { + final String connectionTimeout, final String readTimeout, final RequestCompletionCallback callback, + final EventReporter eventReporter, final NiFiProperties nifiProperties) { if (numThreads <= 0) { throw new IllegalArgumentException("The number of threads must be greater than zero."); } else if (client == null) { @@ -136,7 +140,7 @@ public class ThreadPoolRequestReplicator implements RequestReplicator { this.clusterCoordinator = clusterCoordinator; this.connectionTimeoutMs = (int) FormatUtils.getTimeDuration(connectionTimeout, TimeUnit.MILLISECONDS); this.readTimeoutMs = (int) FormatUtils.getTimeDuration(readTimeout, TimeUnit.MILLISECONDS); - this.responseMerger = new StandardHttpResponseMerger(); + this.responseMerger = new StandardHttpResponseMerger(nifiProperties); this.eventReporter = eventReporter; this.callback = callback;
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/CuratorNodeProtocolSender.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/CuratorNodeProtocolSender.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/CuratorNodeProtocolSender.java index daa3e5c..1847461 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/CuratorNodeProtocolSender.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/node/CuratorNodeProtocolSender.java @@ -14,13 +14,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.apache.nifi.cluster.coordination.node; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; -import java.util.Properties; import org.apache.curator.RetryPolicy; import org.apache.curator.framework.CuratorFramework; @@ -33,6 +31,7 @@ import org.apache.nifi.cluster.protocol.ProtocolException; import org.apache.nifi.cluster.protocol.message.ProtocolMessage; import org.apache.nifi.controller.cluster.ZooKeeperClientConfig; import org.apache.nifi.io.socket.SocketConfiguration; +import org.apache.nifi.util.NiFiProperties; import org.apache.zookeeper.KeeperException.NoNodeException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; @@ -40,19 +39,20 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * Uses Apache Curator to determine the address of the current cluster coordinator + * Uses Apache Curator to determine the address of the current cluster + * coordinator */ public class CuratorNodeProtocolSender extends AbstractNodeProtocolSender { + private static final Logger logger = LoggerFactory.getLogger(CuratorNodeProtocolSender.class); private final String coordinatorPath; private final ZooKeeperClientConfig zkConfig; private InetSocketAddress coordinatorAddress; - - public CuratorNodeProtocolSender(final SocketConfiguration socketConfig, final ProtocolContext<ProtocolMessage> protocolContext, final Properties properties) { + public CuratorNodeProtocolSender(final SocketConfiguration socketConfig, final ProtocolContext<ProtocolMessage> protocolContext, final NiFiProperties nifiProperties) { super(socketConfig, protocolContext); - zkConfig = ZooKeeperClientConfig.createConfig(properties); + zkConfig = ZooKeeperClientConfig.createConfig(nifiProperties); coordinatorPath = zkConfig.resolvePath("cluster/nodes/coordinator"); } @@ -64,7 +64,7 @@ public class CuratorNodeProtocolSender extends AbstractNodeProtocolSender { final RetryPolicy retryPolicy = new RetryNTimes(0, 0); final CuratorFramework curatorClient = CuratorFrameworkFactory.newClient(zkConfig.getConnectString(), - zkConfig.getSessionTimeoutMillis(), zkConfig.getConnectionTimeoutMillis(), retryPolicy); + zkConfig.getSessionTimeoutMillis(), zkConfig.getConnectionTimeoutMillis(), retryPolicy); curatorClient.start(); try { @@ -85,7 +85,7 @@ public class CuratorNodeProtocolSender extends AbstractNodeProtocolSender { final String[] splits = address.split(":"); if (splits.length != 2) { final String message = String.format("Attempted to determine Cluster Coordinator address. Zookeeper indicates " - + "that address is %s, but this is not in the expected format of <hostname>:<port>", address); + + "that address is %s, but this is not in the expected format of <hostname>:<port>", address); logger.error(message); throw new ProtocolException(message); } @@ -101,7 +101,7 @@ public class CuratorNodeProtocolSender extends AbstractNodeProtocolSender { } } catch (final NumberFormatException nfe) { final String message = String.format("Attempted to determine Cluster Coordinator address. Zookeeper indicates " - + "that address is %s, but the port is not a valid port number", address); + + "that address is %s, but the port is not a valid port number", address); logger.error(message); throw new ProtocolException(message); } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/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 ac39dc5..70338c1 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 @@ -14,7 +14,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.apache.nifi.cluster.coordination.node; import java.io.IOException; @@ -25,7 +24,6 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Properties; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; @@ -72,6 +70,7 @@ import org.apache.nifi.controller.cluster.ZooKeeperClientConfig; import org.apache.nifi.events.EventReporter; import org.apache.nifi.reporting.Severity; import org.apache.nifi.services.FlowService; +import org.apache.nifi.util.NiFiProperties; import org.apache.nifi.web.revision.RevisionManager; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; @@ -80,6 +79,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandler, RequestCompletionCallback { + private static final Logger logger = LoggerFactory.getLogger(NodeClusterCoordinator.class); private static final String EVENT_CATEGORY = "Clustering"; @@ -97,6 +97,7 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl private final CuratorFramework curatorClient; private final String nodesPathPrefix; private final String coordinatorPath; + private final NiFiProperties nifiProperties; private volatile FlowService flowService; private volatile boolean connected; @@ -107,18 +108,19 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl private final ConcurrentMap<NodeIdentifier, CircularFifoQueue<NodeEvent>> nodeEvents = new ConcurrentHashMap<>(); public NodeClusterCoordinator(final ClusterCoordinationProtocolSenderListener senderListener, final EventReporter eventReporter, - final ClusterNodeFirewall firewall, final RevisionManager revisionManager, final Properties nifiProperties) { + final ClusterNodeFirewall firewall, final RevisionManager revisionManager, final NiFiProperties nifiProperties) { this.senderListener = senderListener; this.flowService = null; this.eventReporter = eventReporter; this.firewall = firewall; this.revisionManager = revisionManager; + this.nifiProperties = nifiProperties; final RetryPolicy retryPolicy = new RetryNTimes(10, 500); final ZooKeeperClientConfig zkConfig = ZooKeeperClientConfig.createConfig(nifiProperties); curatorClient = CuratorFrameworkFactory.newClient(zkConfig.getConnectString(), - zkConfig.getSessionTimeoutMillis(), zkConfig.getConnectionTimeoutMillis(), retryPolicy); + zkConfig.getSessionTimeoutMillis(), zkConfig.getConnectionTimeoutMillis(), retryPolicy); curatorClient.start(); nodesPathPrefix = zkConfig.resolvePath("cluster/nodes"); @@ -226,12 +228,15 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl } /** - * Attempts to update the nodeStatuses map by changing the value for the given node id from the current status to the new status, as in - * ConcurrentMap.replace(nodeId, currentStatus, newStatus) but with the difference that this method can handle a <code>null</code> value - * for currentStatus + * Attempts to update the nodeStatuses map by changing the value for the + * given node id from the current status to the new status, as in + * ConcurrentMap.replace(nodeId, currentStatus, newStatus) but with the + * difference that this method can handle a <code>null</code> value for + * currentStatus * * @param nodeId the node id - * @param currentStatus the current status, or <code>null</code> if there is no value currently + * @param currentStatus the current status, or <code>null</code> if there is + * no value currently * @param newStatus the new status to set * @return <code>true</code> if the map was updated, false otherwise */ @@ -296,7 +301,6 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl updateNodeStatus(new NodeConnectionStatus(nodeId, NodeConnectionState.CONNECTED, getRoles(nodeId))); } - @Override public void requestNodeDisconnect(final NodeIdentifier nodeId, final DisconnectionCode disconnectionCode, final String explanation) { final Set<NodeIdentifier> connectedNodeIds = getNodeIdentifiers(NodeConnectionState.CONNECTED); @@ -363,7 +367,6 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl return status == null ? null : status.getState(); } - @Override public Map<NodeConnectionState, List<NodeIdentifier>> getConnectionStates() { final Map<NodeConnectionState, List<NodeIdentifier>> connectionStates = new HashMap<>(); @@ -521,18 +524,18 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl } return nodeStatuses.entrySet().stream() - .filter(entry -> statesOfInterest.contains(entry.getValue().getState())) - .map(entry -> entry.getKey()) - .collect(Collectors.toSet()); + .filter(entry -> statesOfInterest.contains(entry.getValue().getState())) + .map(entry -> entry.getKey()) + .collect(Collectors.toSet()); } @Override public NodeIdentifier getPrimaryNode() { return nodeStatuses.values().stream() - .filter(status -> status.getRoles().contains(ClusterRoles.PRIMARY_NODE)) - .findFirst() - .map(status -> status.getNodeIdentifier()) - .orElse(null); + .filter(status -> status.getRoles().contains(ClusterRoles.PRIMARY_NODE)) + .findFirst() + .map(status -> status.getNodeIdentifier()) + .orElse(null); } @Override @@ -582,13 +585,13 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl final Set<NodeIdentifier> connectedNodeIds = getNodeIdentifiers(); final NodeIdentifier electedNodeId = connectedNodeIds.stream() - .filter(nodeId -> nodeId.getSocketAddress().equals(electedNodeHostname) && nodeId.getSocketPort() == electedNodePort) - .findFirst() - .orElse(null); + .filter(nodeId -> nodeId.getSocketAddress().equals(electedNodeHostname) && nodeId.getSocketPort() == electedNodePort) + .findFirst() + .orElse(null); if (electedNodeId == null && warnOnError) { logger.debug("Failed to determine which node is elected active Cluster Coordinator: ZooKeeper reports the address as {}," - + "but there is no node with this address. Will attempt to communicate with node to determine its information", electedNodeAddress); + + "but there is no node with this address. Will attempt to communicate with node to determine its information", electedNodeAddress); try { final NodeConnectionStatus connectionStatus = senderListener.requestNodeConnectionStatus(electedNodeHostname, electedNodePort); @@ -606,7 +609,7 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl } } catch (final Exception e) { logger.warn("Failed to determine which node is elected active Cluster Coordinator: ZooKeeper reports the address as {}, but there is no node with this address. " - + "Attempted to determine the node's information but failed to retrieve its information due to {}", electedNodeAddress, e.toString()); + + "Attempted to determine the node's information but failed to retrieve its information due to {}", electedNodeAddress, e.toString()); if (logger.isDebugEnabled()) { logger.warn("", e); @@ -656,8 +659,9 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl } /** - * Updates the status of the node with the given ID to the given status and returns <code>true</code> - * if successful, <code>false</code> if no node exists with the given ID + * Updates the status of the node with the given ID to the given status and + * returns <code>true</code> if successful, <code>false</code> if no node + * exists with the given ID * * @param status the new status of the node */ @@ -705,7 +709,8 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl * Notifies other nodes that the status of a node changed * * @param updatedStatus the updated status for a node in the cluster - * @param notifyAllNodes if <code>true</code> will notify all nodes. If <code>false</code>, will notify only the cluster coordinator + * @param notifyAllNodes if <code>true</code> will notify all nodes. If + * <code>false</code>, will notify only the cluster coordinator */ void notifyOthersOfNodeStatusChange(final NodeConnectionStatus updatedStatus, final boolean notifyAllNodes, final boolean waitForCoordinator) { // If this node is the active cluster coordinator, then we are going to replicate to all nodes. @@ -770,7 +775,7 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl Thread.sleep(100L); } catch (final InterruptedException ie) { logger.info("Could not send Reconnection request to {} because thread was " - + "interrupted before FlowService was made available", request.getNodeId()); + + "interrupted before FlowService was made available", request.getNodeId()); Thread.currentThread().interrupt(); return; } @@ -797,7 +802,7 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl } catch (final Exception e) { logger.warn("Problem encountered issuing reconnection request to node " + request.getNodeId(), e); eventReporter.reportEvent(Severity.WARNING, EVENT_CATEGORY, "Problem encountered issuing reconnection request to node " - + request.getNodeId() + " due to: " + e); + + request.getNodeId() + " due to: " + e); } try { @@ -810,7 +815,7 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl // We failed to reconnect too many times. We must now mark node as disconnected. if (NodeConnectionState.CONNECTING == getConnectionState(request.getNodeId())) { requestNodeDisconnect(request.getNodeId(), DisconnectionCode.UNABLE_TO_COMMUNICATE, - "Attempted to request that node reconnect to cluster but could not communicate with node"); + "Attempted to request that node reconnect to cluster but could not communicate with node"); } } }, "Reconnect " + request.getNodeId()); @@ -944,10 +949,10 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl } else { // there is a node with that ID and it's a different node resolvedNodeId = new NodeIdentifier(UUID.randomUUID().toString(), proposedIdentifier.getApiAddress(), proposedIdentifier.getApiPort(), - proposedIdentifier.getSocketAddress(), proposedIdentifier.getSocketPort(), proposedIdentifier.getSiteToSiteAddress(), - proposedIdentifier.getSiteToSitePort(), proposedIdentifier.getSiteToSiteHttpApiPort(), proposedIdentifier.isSiteToSiteSecure()); + proposedIdentifier.getSocketAddress(), proposedIdentifier.getSocketPort(), proposedIdentifier.getSiteToSiteAddress(), + proposedIdentifier.getSiteToSitePort(), proposedIdentifier.getSiteToSiteHttpApiPort(), proposedIdentifier.isSiteToSiteSecure()); logger.debug("A node already exists with ID {}. Proposed Node Identifier was {}; existing Node Identifier is {}; Resolved Node Identifier is {}", - proposedIdentifier.getId(), proposedIdentifier, getNodeIdentifier(proposedIdentifier.getId()), resolvedNodeId); + proposedIdentifier.getId(), proposedIdentifier, getNodeIdentifier(proposedIdentifier.getId()), resolvedNodeId); } return resolvedNodeId; @@ -989,7 +994,7 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl dataFlow = flowService.createDataFlow(); } catch (final IOException ioe) { logger.error("Unable to obtain current dataflow from FlowService in order to provide the flow to " - + resolvedNodeIdentifier + ". Will tell node to try again later", ioe); + + resolvedNodeIdentifier + ". Will tell node to try again later", ioe); } } @@ -998,37 +1003,37 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl // the flow management service a chance to retrieve a current flow final int tryAgainSeconds = 5; addNodeEvent(resolvedNodeIdentifier, Severity.WARNING, "Connection requested from node, but manager was unable to obtain current flow. " - + "Instructing node to try again in " + tryAgainSeconds + " seconds."); + + "Instructing node to try again in " + tryAgainSeconds + " seconds."); // return try later response return new ConnectionResponse(tryAgainSeconds); } return new ConnectionResponse(resolvedNodeIdentifier, dataFlow, instanceId, new ArrayList<>(nodeStatuses.values()), - revisionManager.getAllRevisions().stream().map(rev -> ComponentRevision.fromRevision(rev)).collect(Collectors.toList())); + revisionManager.getAllRevisions().stream().map(rev -> ComponentRevision.fromRevision(rev)).collect(Collectors.toList())); } private NodeIdentifier addRequestorDn(final NodeIdentifier nodeId, final String dn) { return new NodeIdentifier(nodeId.getId(), nodeId.getApiAddress(), nodeId.getApiPort(), - nodeId.getSocketAddress(), nodeId.getSocketPort(), - nodeId.getSiteToSiteAddress(), nodeId.getSiteToSitePort(), - nodeId.getSiteToSiteHttpApiPort(), nodeId.isSiteToSiteSecure(), dn); + nodeId.getSocketAddress(), nodeId.getSocketPort(), + nodeId.getSiteToSiteAddress(), nodeId.getSiteToSitePort(), + nodeId.getSiteToSiteHttpApiPort(), nodeId.isSiteToSiteSecure(), dn); } @Override public boolean canHandle(final ProtocolMessage msg) { return MessageType.CONNECTION_REQUEST == msg.getType() || MessageType.NODE_STATUS_CHANGE == msg.getType() - || MessageType.NODE_CONNECTION_STATUS_REQUEST == msg.getType(); + || MessageType.NODE_CONNECTION_STATUS_REQUEST == msg.getType(); } private boolean isMutableRequest(final String method) { return "DELETE".equalsIgnoreCase(method) || "POST".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method); } - /** - * Callback that is called after an HTTP Request has been replicated to nodes in the cluster. - * This allows us to disconnect nodes that did not complete the request, if applicable. + * Callback that is called after an HTTP Request has been replicated to + * nodes in the cluster. This allows us to disconnect nodes that did not + * complete the request, if applicable. */ @Override public void afterRequest(final String uriPath, final String method, final Set<NodeResponse> nodeResponses) { @@ -1047,7 +1052,7 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl * state even if they had problems handling the request. */ if (mutableRequest) { - final HttpResponseMerger responseMerger = new StandardHttpResponseMerger(); + final HttpResponseMerger responseMerger = new StandardHttpResponseMerger(nifiProperties); final Set<NodeResponse> problematicNodeResponses = responseMerger.getProblematicNodeResponses(nodeResponses); // all nodes failed @@ -1055,7 +1060,7 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl // some nodes had a problematic response because of a missing counter, ensure the are not disconnected final boolean someNodesFailedMissingCounter = !problematicNodeResponses.isEmpty() - && problematicNodeResponses.size() < nodeResponses.size() && isMissingCounter(problematicNodeResponses, uriPath); + && problematicNodeResponses.size() < nodeResponses.size() && isMissingCounter(problematicNodeResponses, uriPath); // ensure nodes stay connected in certain scenarios if (allNodesFailed) { @@ -1079,12 +1084,15 @@ public class NodeClusterCoordinator implements ClusterCoordinator, ProtocolHandl } /** - * Determines if all problematic responses were due to 404 NOT_FOUND. Assumes that problematicNodeResponses is not empty and is not comprised of responses from all nodes in the cluster (at least - * one node contained the counter in question). + * Determines if all problematic responses were due to 404 NOT_FOUND. + * Assumes that problematicNodeResponses is not empty and is not comprised + * of responses from all nodes in the cluster (at least one node contained + * the counter in question). * * @param problematicNodeResponses The problematic node responses * @param uriPath The path of the URI for the request - * @return Whether all problematic node responses were due to a missing counter + * @return Whether all problematic node responses were due to a missing + * counter */ private boolean isMissingCounter(final Set<NodeResponse> problematicNodeResponses, final String uriPath) { if (COUNTER_URI_PATTERN.matcher(uriPath).matches()) { http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/spring/ThreadPoolRequestReplicatorFactoryBean.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/spring/ThreadPoolRequestReplicatorFactoryBean.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/spring/ThreadPoolRequestReplicatorFactoryBean.java index fc0eaf2..31c3b1d 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/spring/ThreadPoolRequestReplicatorFactoryBean.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/spring/ThreadPoolRequestReplicatorFactoryBean.java @@ -33,24 +33,24 @@ import org.springframework.context.ApplicationContextAware; public class ThreadPoolRequestReplicatorFactoryBean implements FactoryBean<ThreadPoolRequestReplicator>, ApplicationContextAware { private ApplicationContext applicationContext; - private NiFiProperties properties; + private NiFiProperties nifiProperties; private ThreadPoolRequestReplicator replicator = null; @Override public ThreadPoolRequestReplicator getObject() throws Exception { - if (replicator == null && properties.isNode()) { + if (replicator == null && nifiProperties.isNode()) { final EventReporter eventReporter = applicationContext.getBean("eventReporter", EventReporter.class); final ClusterCoordinator clusterCoordinator = applicationContext.getBean("clusterCoordinator", ClusterCoordinator.class); final RequestCompletionCallback requestCompletionCallback = applicationContext.getBean("clusterCoordinator", RequestCompletionCallback.class); - final int numThreads = properties.getClusterNodeProtocolThreads(); - final Client jerseyClient = WebUtils.createClient(new DefaultClientConfig(), SslContextFactory.createSslContext(properties)); - final String connectionTimeout = properties.getClusterNodeConnectionTimeout(); - final String readTimeout = properties.getClusterNodeReadTimeout(); + final int numThreads = nifiProperties.getClusterNodeProtocolThreads(); + final Client jerseyClient = WebUtils.createClient(new DefaultClientConfig(), SslContextFactory.createSslContext(nifiProperties)); + final String connectionTimeout = nifiProperties.getClusterNodeConnectionTimeout(); + final String readTimeout = nifiProperties.getClusterNodeReadTimeout(); replicator = new ThreadPoolRequestReplicator(numThreads, jerseyClient, clusterCoordinator, - connectionTimeout, readTimeout, requestCompletionCallback, eventReporter); + connectionTimeout, readTimeout, requestCompletionCallback, eventReporter, nifiProperties); } return replicator; @@ -71,8 +71,8 @@ public class ThreadPoolRequestReplicatorFactoryBean implements FactoryBean<Threa this.applicationContext = applicationContext; } - public void setProperties(NiFiProperties properties) { - this.properties = properties; + public void setProperties(final NiFiProperties properties) { + this.nifiProperties = properties; } } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/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 5086dc0..2307538 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 @@ -27,7 +27,6 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Properties; import java.util.Set; import java.util.UUID; import java.util.stream.Collectors; @@ -52,6 +51,7 @@ public class TestAbstractHeartbeatMonitor { @Before public void setup() throws Exception { + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/conf/nifi.properties"); nodeId = new NodeIdentifier(UUID.randomUUID().toString(), "localhost", 9999, "localhost", 8888, "localhost", null, null, false); } @@ -179,10 +179,10 @@ public class TestAbstractHeartbeatMonitor { return monitor; } - private Properties createProperties() { - final Properties properties = new Properties(); - properties.setProperty(NiFiProperties.CLUSTER_PROTOCOL_HEARTBEAT_INTERVAL, "10 ms"); - return properties; + private NiFiProperties createProperties() { + final Map<String, String> addProps = new HashMap<>(); + addProps.put(NiFiProperties.CLUSTER_PROTOCOL_HEARTBEAT_INTERVAL, "10 ms"); + return NiFiProperties.createBasicNiFiProperties(null, addProps); } private static class ClusterCoordinatorAdapter implements ClusterCoordinator { @@ -328,8 +328,8 @@ public class TestAbstractHeartbeatMonitor { private Map<NodeIdentifier, NodeHeartbeat> heartbeats = new HashMap<>(); private final Object mutex = new Object(); - public TestFriendlyHeartbeatMonitor(ClusterCoordinator clusterCoordinator, Properties properties) { - super(clusterCoordinator, properties); + public TestFriendlyHeartbeatMonitor(ClusterCoordinator clusterCoordinator, NiFiProperties nifiProperties) { + super(clusterCoordinator, nifiProperties); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMergerSpec.groovy ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMergerSpec.groovy b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMergerSpec.groovy index bd9b265..03aa08a 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMergerSpec.groovy +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMergerSpec.groovy @@ -46,7 +46,8 @@ import spock.lang.Unroll class StandardHttpResponseMergerSpec extends Specification { def setup() { - System.setProperty NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/conf/nifi.properties" + def propFile = StandardHttpResponseMergerSpec.class.getResource("/conf/nifi.properties").getFile() + System.setProperty NiFiProperties.PROPERTIES_FILE_PATH, propFile } def cleanup() { @@ -55,7 +56,7 @@ class StandardHttpResponseMergerSpec extends Specification { def "MergeResponses: mixed HTTP GET response statuses, expecting #expectedStatus"() { given: - def responseMerger = new StandardHttpResponseMerger() + def responseMerger = new StandardHttpResponseMerger(NiFiProperties.createBasicNiFiProperties(null,null)) def requestUri = new URI('http://server/resource') def requestId = UUID.randomUUID().toString() def Map<ClientResponse, Map<String, Integer>> mockToRequestEntity = [:] @@ -93,7 +94,7 @@ class StandardHttpResponseMergerSpec extends Specification { mapper.setSerializationConfig(serializationConfig.withSerializationInclusion(JsonSerialize.Inclusion.NON_NULL).withAnnotationIntrospector(jaxbIntrospector)); and: "setup of the data to be used in the test" - def responseMerger = new StandardHttpResponseMerger() + def responseMerger = new StandardHttpResponseMerger(NiFiProperties.createBasicNiFiProperties(null,null)) def requestUri = new URI("http://server/$requestUriPart";) def requestId = UUID.randomUUID().toString() def Map<ClientResponse, Object> mockToRequestEntity = [:] http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/StatusHistoryEndpointMergerSpec.groovy ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/StatusHistoryEndpointMergerSpec.groovy b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/StatusHistoryEndpointMergerSpec.groovy index 69dd82a..350269d 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/StatusHistoryEndpointMergerSpec.groovy +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/endpoints/StatusHistoryEndpointMergerSpec.groovy @@ -32,7 +32,8 @@ import spock.lang.Unroll class StatusHistoryEndpointMergerSpec extends Specification { def setup() { - System.setProperty NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/conf/nifi.properties" + def propFile = StatusHistoryEndpointMergerSpec.class.getResource("/conf/nifi.properties").getFile() + System.setProperty NiFiProperties.PROPERTIES_FILE_PATH, propFile } def cleanup() { @@ -48,7 +49,7 @@ class StatusHistoryEndpointMergerSpec extends Specification { mapper.setSerializationConfig(serializationConfig.withSerializationInclusion(JsonSerialize.Inclusion.NON_NULL).withAnnotationIntrospector(jaxbIntrospector)); and: "setup of the data to be used in the test" - def merger = new StatusHistoryEndpointMerger() + def merger = new StatusHistoryEndpointMerger(2) def requestUri = new URI("http://server/$requestUriPart";) def requestId = UUID.randomUUID().toString() def Map<ClientResponse, Object> mockToRequestEntity = [:] http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestThreadPoolRequestReplicator.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestThreadPoolRequestReplicator.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestThreadPoolRequestReplicator.java index 5eac846..ebefce2 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestThreadPoolRequestReplicator.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/coordination/http/replication/TestThreadPoolRequestReplicator.java @@ -14,7 +14,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.apache.nifi.cluster.coordination.http.replication; import static org.junit.Assert.assertEquals; @@ -75,8 +74,9 @@ public class TestThreadPoolRequestReplicator { } /** - * If we replicate a request, whenever we obtain the merged response from the AsyncClusterResponse object, - * the response should no longer be available and should be cleared from internal state. This test is to + * If we replicate a request, whenever we obtain the merged response from + * the AsyncClusterResponse object, the response should no longer be + * available and should be cleared from internal state. This test is to * verify that this behavior occurs. */ @Test @@ -105,7 +105,6 @@ public class TestThreadPoolRequestReplicator { }); } - @Test(timeout = 15000) public void testLongWaitForResponse() { withReplicator(replicator -> { @@ -132,7 +131,7 @@ public class TestThreadPoolRequestReplicator { assertTrue(response.isComplete()); assertNotNull(response.getMergedResponse()); assertNull(replicator.getClusterResponse(response.getRequestIdentifier())); - } , Status.OK, 1000, new ClientHandlerException(new SocketTimeoutException())); + }, Status.OK, 1000, new ClientHandlerException(new SocketTimeoutException())); } @Test(timeout = 15000) @@ -153,10 +152,9 @@ public class TestThreadPoolRequestReplicator { final AsyncClusterResponse response = replicator.replicate(nodeIds, HttpMethod.GET, uri, entity, new HashMap<>(), true, true); assertNotNull(response.awaitMergedResponse(1, TimeUnit.SECONDS)); - } , null, 0L, new IllegalArgumentException("Exception created for unit test")); + }, null, 0L, new IllegalArgumentException("Exception created for unit test")); } - @Test(timeout = 15000) public void testMultipleRequestWithTwoPhaseCommit() { final Set<NodeIdentifier> nodeIds = new HashSet<>(); @@ -167,7 +165,8 @@ public class TestThreadPoolRequestReplicator { Mockito.when(coordinator.getConnectionStatus(Mockito.any(NodeIdentifier.class))).thenReturn(new NodeConnectionStatus(nodeId, NodeConnectionState.CONNECTED, Collections.emptySet())); final AtomicInteger requestCount = new AtomicInteger(0); - final ThreadPoolRequestReplicator replicator = new ThreadPoolRequestReplicator(2, new Client(), coordinator, "1 sec", "1 sec", null, null) { + final ThreadPoolRequestReplicator replicator + = new ThreadPoolRequestReplicator(2, new Client(), coordinator, "1 sec", "1 sec", null, null, NiFiProperties.createBasicNiFiProperties(null, null)) { @Override protected NodeResponse replicateRequest(final WebResource.Builder resourceBuilder, final NodeIdentifier nodeId, final String method, final URI uri, final String requestId) { // the resource builder will not expose its headers to us, so we are using Mockito's Whitebox class to extract them. @@ -191,7 +190,7 @@ public class TestThreadPoolRequestReplicator { try { final AsyncClusterResponse clusterResponse = replicator.replicate(nodeIds, HttpMethod.POST, - new URI("http://localhost:80/processors/1";), new ProcessorEntity(), new HashMap<>(), true, true); + new URI("http://localhost:80/processors/1";), new ProcessorEntity(), new HashMap<>(), true, true); clusterResponse.awaitMergedResponse(); // Ensure that we received two requests - the first should contain the X-NcmExpects header; the second should not. @@ -233,7 +232,8 @@ public class TestThreadPoolRequestReplicator { nodeMap.put(NodeConnectionState.CONNECTING, otherState); Mockito.when(coordinator.getConnectionStates()).thenReturn(nodeMap); - final ThreadPoolRequestReplicator replicator = new ThreadPoolRequestReplicator(2, new Client(), coordinator, "1 sec", "1 sec", null, null) { + final ThreadPoolRequestReplicator replicator + = new ThreadPoolRequestReplicator(2, new Client(), coordinator, "1 sec", "1 sec", null, null, NiFiProperties.createBasicNiFiProperties(null, null)) { @Override public AsyncClusterResponse replicate(Set<NodeIdentifier> nodeIds, String method, URI uri, Object entity, Map<String, String> headers, boolean indicateReplicated, boolean verify) { @@ -278,7 +278,6 @@ public class TestThreadPoolRequestReplicator { } } - @Test(timeout = 15000) public void testOneNodeRejectsTwoPhaseCommit() { final Set<NodeIdentifier> nodeIds = new HashSet<>(); @@ -287,7 +286,8 @@ public class TestThreadPoolRequestReplicator { final ClusterCoordinator coordinator = createClusterCoordinator(); final AtomicInteger requestCount = new AtomicInteger(0); - final ThreadPoolRequestReplicator replicator = new ThreadPoolRequestReplicator(2, new Client(), coordinator, "1 sec", "1 sec", null, null) { + final ThreadPoolRequestReplicator replicator + = new ThreadPoolRequestReplicator(2, new Client(), coordinator, "1 sec", "1 sec", null, null, NiFiProperties.createBasicNiFiProperties(null, null)) { @Override protected NodeResponse replicateRequest(final WebResource.Builder resourceBuilder, final NodeIdentifier nodeId, final String method, final URI uri, final String requestId) { // the resource builder will not expose its headers to us, so we are using Mockito's Whitebox class to extract them. @@ -309,7 +309,7 @@ public class TestThreadPoolRequestReplicator { try { final AsyncClusterResponse clusterResponse = replicator.replicate(nodeIds, HttpMethod.POST, - new URI("http://localhost:80/processors/1";), new ProcessorEntity(), new HashMap<>(), true, true); + new URI("http://localhost:80/processors/1";), new ProcessorEntity(), new HashMap<>(), true, true); clusterResponse.awaitMergedResponse(); Assert.fail("Expected to get an IllegalClusterStateException but did not"); @@ -322,15 +322,14 @@ public class TestThreadPoolRequestReplicator { } } - - private void withReplicator(final WithReplicator function) { withReplicator(function, ClientResponse.Status.OK, 0L, null); } private void withReplicator(final WithReplicator function, final Status status, final long delayMillis, final RuntimeException failure) { final ClusterCoordinator coordinator = createClusterCoordinator(); - final ThreadPoolRequestReplicator replicator = new ThreadPoolRequestReplicator(2, new Client(), coordinator, "1 sec", "1 sec", null, null) { + final ThreadPoolRequestReplicator replicator + = new ThreadPoolRequestReplicator(2, new Client(), coordinator, "1 sec", "1 sec", null, null, NiFiProperties.createBasicNiFiProperties(null, null)) { @Override protected NodeResponse replicateRequest(final WebResource.Builder resourceBuilder, final NodeIdentifier nodeId, final String method, final URI uri, final String requestId) { if (delayMillis > 0L) { @@ -362,6 +361,7 @@ public class TestThreadPoolRequestReplicator { } private interface WithReplicator { + void withReplicator(ThreadPoolRequestReplicator replicator) throws Exception; } } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/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 2f034b3..aaa9dca 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 @@ -14,7 +14,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.apache.nifi.cluster.coordination.node; import static org.junit.Assert.assertEquals; @@ -27,9 +26,9 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Properties; import java.util.Set; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; @@ -47,6 +46,7 @@ import org.apache.nifi.cluster.protocol.message.ProtocolMessage; import org.apache.nifi.cluster.protocol.message.ReconnectionRequestMessage; import org.apache.nifi.events.EventReporter; import org.apache.nifi.services.FlowService; +import org.apache.nifi.util.NiFiProperties; import org.apache.nifi.web.revision.RevisionManager; import org.junit.Assert; import org.junit.Before; @@ -56,18 +56,21 @@ import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; public class TestNodeClusterCoordinator { + private NodeClusterCoordinator coordinator; private ClusterCoordinationProtocolSenderListener senderListener; private List<NodeConnectionStatus> nodeStatuses; - private Properties createProperties() { - final Properties props = new Properties(); - props.put("nifi.zookeeper.connect.string", "localhost:2181"); - return props; + private NiFiProperties createProperties() { + final Map<String,String> addProps = new HashMap<>(); + addProps.put("nifi.zookeeper.connect.string", "localhost:2181"); + return NiFiProperties.createBasicNiFiProperties(null, addProps); } @Before public void setup() throws IOException { + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/conf/nifi.properties"); + senderListener = Mockito.mock(ClusterCoordinationProtocolSenderListener.class); nodeStatuses = Collections.synchronizedList(new ArrayList<>()); @@ -113,7 +116,7 @@ public class TestNodeClusterCoordinator { assertNotNull(statuses); assertEquals(6, statuses.size()); final Map<NodeIdentifier, NodeConnectionStatus> statusMap = statuses.stream().collect( - Collectors.toMap(status -> status.getNodeIdentifier(), status -> status)); + Collectors.toMap(status -> status.getNodeIdentifier(), status -> status)); assertEquals(DisconnectionCode.LACK_OF_HEARTBEAT, statusMap.get(createNodeId(1)).getDisconnectCode()); assertEquals(NodeConnectionState.DISCONNECTING, statusMap.get(createNodeId(2)).getState()); @@ -258,7 +261,6 @@ public class TestNodeClusterCoordinator { assertEquals("Unit Test", statusChange.getDisconnectReason()); } - @Test public void testGetConnectionStates() throws IOException { // Add a disconnected node @@ -316,7 +318,6 @@ public class TestNodeClusterCoordinator { assertTrue(disconnectedIds.contains(createNodeId(1))); } - @Test(timeout = 5000) public void testRequestNodeDisconnect() throws InterruptedException { // Add a connected node @@ -341,7 +342,6 @@ public class TestNodeClusterCoordinator { assertEquals(NodeConnectionState.DISCONNECTED, status.getState()); } - @Test(timeout = 5000) public void testCannotDisconnectLastNode() throws InterruptedException { // Add a connected node @@ -369,7 +369,6 @@ public class TestNodeClusterCoordinator { coordinator.requestNodeDisconnect(nodeId2, DisconnectionCode.USER_DISCONNECTED, "Unit Test"); } - @Test(timeout = 5000) public void testUpdateNodeStatusOutOfOrder() throws InterruptedException { // Add a connected node @@ -386,7 +385,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, null); final NodeStatusChangeMessage msg = new NodeStatusChangeMessage(); msg.setNodeId(nodeId1); msg.setNodeConnectionStatus(oldStatus); @@ -452,7 +451,6 @@ public class TestNodeClusterCoordinator { assertEquals(Collections.singleton(ClusterRoles.PRIMARY_NODE), id2Msg.getRoles()); } - @Test public void testProposedIdentifierResolvedIfConflict() { final NodeIdentifier id1 = new NodeIdentifier("1234", "localhost", 8000, "localhost", 9000, "localhost", 10000, 11000, false); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/integration/Cluster.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/integration/Cluster.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/integration/Cluster.java index dbd8c00..5809625 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/integration/Cluster.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/integration/Cluster.java @@ -18,7 +18,9 @@ package org.apache.nifi.cluster.integration; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; +import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; @@ -91,11 +93,11 @@ public class Cluster { public Node createNode() { - NiFiProperties.getInstance().setProperty(NiFiProperties.ZOOKEEPER_CONNECT_STRING, getZooKeeperConnectString()); - NiFiProperties.getInstance().setProperty(NiFiProperties.CLUSTER_IS_NODE, "true"); + final Map<String, String> addProps = new HashMap<>(); + addProps.put(NiFiProperties.ZOOKEEPER_CONNECT_STRING, getZooKeeperConnectString()); + addProps.put(NiFiProperties.CLUSTER_IS_NODE, "true"); - final NiFiProperties properties = NiFiProperties.getInstance().copy(); - final Node node = new Node(properties); + final Node node = new Node(NiFiProperties.createBasicNiFiProperties("src/test/resources/conf/nifi.properties", addProps)); node.start(); nodes.add(node); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/integration/Node.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/integration/Node.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/integration/Node.java index 5bfe83c..899f312 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/integration/Node.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/integration/Node.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.net.ServerSocket; import java.util.ArrayList; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Set; import java.util.UUID; @@ -90,10 +91,26 @@ public class Node { public Node(final NodeIdentifier nodeId, final NiFiProperties properties) { this.nodeId = nodeId; - this.nodeProperties = properties; + this.nodeProperties = new NiFiProperties() { + @Override + public String getProperty(String key) { + if(key.equals(NiFiProperties.CLUSTER_NODE_PROTOCOL_PORT)){ + return String.valueOf(nodeId.getSocketPort()); + }else if(key.equals(NiFiProperties.WEB_HTTP_PORT)){ + return String.valueOf(nodeId.getApiPort()); + }else { + return properties.getProperty(key); + } + } - nodeProperties.setProperty(NiFiProperties.CLUSTER_NODE_PROTOCOL_PORT, String.valueOf(nodeId.getSocketPort())); - nodeProperties.setProperty(NiFiProperties.WEB_HTTP_PORT, String.valueOf(nodeId.getApiPort())); + @Override + public Set<String> getPropertyKeys() { + final Set<String> keys = new HashSet<>(properties.getPropertyKeys()); + keys.add(NiFiProperties.CLUSTER_NODE_PROTOCOL_PORT); + keys.add(NiFiProperties.WEB_HTTP_PORT); + return keys; + } + }; revisionManager = Mockito.mock(RevisionManager.class); Mockito.when(revisionManager.getAllRevisions()).thenReturn(Collections.<Revision> emptyList()); @@ -110,7 +127,7 @@ public class Node { final HeartbeatMonitor heartbeatMonitor = createHeartbeatMonitor(); flowController = FlowController.createClusteredInstance(Mockito.mock(FlowFileEventRepository.class), nodeProperties, - null, null, StringEncryptor.createEncryptor(), protocolSender, Mockito.mock(BulletinRepository.class), clusterCoordinator, heartbeatMonitor, VariableRegistry.EMPTY_REGISTRY); + null, null, StringEncryptor.createEncryptor(nodeProperties), protocolSender, Mockito.mock(BulletinRepository.class), clusterCoordinator, heartbeatMonitor, VariableRegistry.EMPTY_REGISTRY); try { flowController.initializeFlow(); @@ -123,7 +140,7 @@ public class Node { flowController.getStateManagerProvider().getStateManager("Cluster Node Configuration").setState(Collections.singletonMap("Node UUID", nodeId.getId()), Scope.LOCAL); flowService = StandardFlowService.createClusteredInstance(flowController, nodeProperties, senderListener, clusterCoordinator, - StringEncryptor.createEncryptor(), revisionManager, Mockito.mock(Authorizer.class)); + StringEncryptor.createEncryptor(nodeProperties), revisionManager, Mockito.mock(Authorizer.class)); flowService.start(); flowService.load(null); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/remote/protocol/ServerProtocol.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/remote/protocol/ServerProtocol.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/remote/protocol/ServerProtocol.java index 4f86001..4366068 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/remote/protocol/ServerProtocol.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/remote/protocol/ServerProtocol.java @@ -134,10 +134,20 @@ public interface ServerProtocol extends VersionedRemoteResource { * * @param peer peer * @param clusterNodeInfo the cluster information + * @param remoteInputHost the remote input host + * @param remoteInputPort the remote input port + * @param remoteInputHttpPort the remote input http port + * @param isSiteToSiteSecure whether site to site is secure * * @throws java.io.IOException ioe */ - void sendPeerList(Peer peer, Optional<ClusterNodeInformation> clusterNodeInfo) throws IOException; + void sendPeerList( + Peer peer, + Optional<ClusterNodeInformation> clusterNodeInfo, + String remoteInputHost, + int remoteInputPort, + int remoteInputHttpPort, + boolean isSiteToSiteSecure) throws IOException; void shutdown(Peer peer); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/connectable/StandardConnection.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/connectable/StandardConnection.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/connectable/StandardConnection.java index 5ad9a3c..2a0f0de 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/connectable/StandardConnection.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/connectable/StandardConnection.java @@ -40,7 +40,6 @@ import org.apache.nifi.groups.ProcessGroup; import org.apache.nifi.processor.FlowFileFilter; import org.apache.nifi.processor.Relationship; import org.apache.nifi.provenance.ProvenanceEventRepository; -import org.apache.nifi.util.NiFiProperties; import java.util.ArrayList; import java.util.Collection; @@ -55,7 +54,9 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; /** - * Models a connection between connectable components. A connection may contain one or more relationships that map the source component to the destination component. + * Models a connection between connectable components. A connection may contain + * one or more relationships that map the source component to the destination + * component. */ public final class StandardConnection implements Connection { @@ -82,7 +83,7 @@ public final class StandardConnection implements Connection { relationships = new AtomicReference<>(Collections.unmodifiableCollection(builder.relationships)); scheduler = builder.scheduler; flowFileQueue = new StandardFlowFileQueue(id, this, builder.flowFileRepository, builder.provenanceRepository, builder.resourceClaimManager, - scheduler, builder.swapManager, builder.eventReporter, NiFiProperties.getInstance().getQueueSwapThreshold()); + scheduler, builder.swapManager, builder.eventReporter, builder.queueSwapThreshold); hashCode = new HashCodeBuilder(7, 67).append(id).toHashCode(); } @@ -307,8 +308,10 @@ public final class StandardConnection implements Connection { } /** - * Gives this Connection ownership of the given FlowFile and allows the Connection to hold on to the FlowFile but NOT provide the FlowFile to consumers. This allows us to ensure that the - * Connection is not deleted during the middle of a Session commit. + * Gives this Connection ownership of the given FlowFile and allows the + * Connection to hold on to the FlowFile but NOT provide the FlowFile to + * consumers. This allows us to ensure that the Connection is not deleted + * during the middle of a Session commit. * * @param flowFile to add */ @@ -338,6 +341,7 @@ public final class StandardConnection implements Connection { private FlowFileRepository flowFileRepository; private ProvenanceEventRepository provenanceRepository; private ResourceClaimManager resourceClaimManager; + private int queueSwapThreshold; public Builder(final ProcessScheduler scheduler) { this.scheduler = scheduler; @@ -409,6 +413,11 @@ public final class StandardConnection implements Connection { return this; } + public Builder queueSwapThreshold(final int queueSwapThreshold) { + this.queueSwapThreshold = queueSwapThreshold; + return this; + } + public StandardConnection build() { if (source == null) { throw new IllegalStateException("Cannot build a Connection without a Source"); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FileSystemSwapManager.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FileSystemSwapManager.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FileSystemSwapManager.java index a4c267c..a61d7fe 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FileSystemSwapManager.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FileSystemSwapManager.java @@ -63,7 +63,8 @@ import org.slf4j.LoggerFactory; /** * <p> - * An implementation of the {@link FlowFileSwapManager} that swaps FlowFiles to/from local disk + * An implementation of the {@link FlowFileSwapManager} that swaps FlowFiles + * to/from local disk * </p> */ public class FileSystemSwapManager implements FlowFileSwapManager { @@ -83,9 +84,15 @@ public class FileSystemSwapManager implements FlowFileSwapManager { private EventReporter eventReporter; private ResourceClaimManager claimManager; + /** + * Default no args constructor for service loading only. + */ public FileSystemSwapManager() { - final NiFiProperties properties = NiFiProperties.getInstance(); - final Path flowFileRepoPath = properties.getFlowFileRepositoryPath(); + storageDirectory = null; + } + + public FileSystemSwapManager(final NiFiProperties nifiProperties) { + final Path flowFileRepoPath = nifiProperties.getFlowFileRepositoryPath(); this.storageDirectory = flowFileRepoPath.resolve("swap").toFile(); if (!storageDirectory.exists() && !storageDirectory.mkdirs()) { @@ -93,7 +100,6 @@ public class FileSystemSwapManager implements FlowFileSwapManager { } } - @Override public synchronized void initialize(final SwapManagerInitializationContext initializationContext) { this.claimManager = initializationContext.getResourceClaimManager(); @@ -129,7 +135,6 @@ public class FileSystemSwapManager implements FlowFileSwapManager { return swapLocation; } - @Override public SwapContents swapIn(final String swapLocation, final FlowFileQueue flowFileQueue) throws IOException { final File swapFile = new File(swapLocation); @@ -152,15 +157,14 @@ public class FileSystemSwapManager implements FlowFileSwapManager { final SwapContents swapContents; try (final InputStream fis = new FileInputStream(swapFile); - final InputStream bis = new BufferedInputStream(fis); - final DataInputStream in = new DataInputStream(bis)) { + final InputStream bis = new BufferedInputStream(fis); + final DataInputStream in = new DataInputStream(bis)) { swapContents = deserializeFlowFiles(in, swapLocation, flowFileQueue, claimManager); } return swapContents; } - @Override public void purge() { final File[] swapFiles = storageDirectory.listFiles(new FilenameFilter() { @@ -177,7 +181,6 @@ public class FileSystemSwapManager implements FlowFileSwapManager { } } - @Override public List<String> recoverSwapLocations(final FlowFileQueue flowFileQueue) throws IOException { final File[] swapFiles = storageDirectory.listFiles(new FilenameFilter() { @@ -217,13 +220,13 @@ public class FileSystemSwapManager implements FlowFileSwapManager { // Read the queue identifier from the swap file to check if the swap file is for this queue try (final InputStream fis = new FileInputStream(swapFile); - final InputStream bufferedIn = new BufferedInputStream(fis); - final DataInputStream in = new DataInputStream(bufferedIn)) { + final InputStream bufferedIn = new BufferedInputStream(fis); + final DataInputStream in = new DataInputStream(bufferedIn)) { final int swapEncodingVersion = in.readInt(); if (swapEncodingVersion > SWAP_ENCODING_VERSION) { final String errMsg = "Cannot swap FlowFiles in from " + swapFile + " because the encoding version is " - + swapEncodingVersion + ", which is too new (expecting " + SWAP_ENCODING_VERSION + " or less)"; + + swapEncodingVersion + ", which is too new (expecting " + SWAP_ENCODING_VERSION + " or less)"; eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, errMsg); throw new IOException(errMsg); @@ -246,13 +249,13 @@ public class FileSystemSwapManager implements FlowFileSwapManager { // read record from disk via the swap file try (final InputStream fis = new FileInputStream(swapFile); - final InputStream bufferedIn = new BufferedInputStream(fis); - final DataInputStream in = new DataInputStream(bufferedIn)) { + final InputStream bufferedIn = new BufferedInputStream(fis); + final DataInputStream in = new DataInputStream(bufferedIn)) { final int swapEncodingVersion = in.readInt(); if (swapEncodingVersion > SWAP_ENCODING_VERSION) { final String errMsg = "Cannot swap FlowFiles in from " + swapFile + " because the encoding version is " - + swapEncodingVersion + ", which is too new (expecting " + SWAP_ENCODING_VERSION + " or less)"; + + swapEncodingVersion + ", which is too new (expecting " + SWAP_ENCODING_VERSION + " or less)"; eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, errMsg); throw new IOException(errMsg); @@ -348,7 +351,7 @@ public class FileSystemSwapManager implements FlowFileSwapManager { out.flush(); } - logger.info("Successfully swapped out {} FlowFiles from {} to Swap File {}", new Object[] {toSwap.size(), queue, swapLocation}); + logger.info("Successfully swapped out {} FlowFiles from {} to Swap File {}", new Object[]{toSwap.size(), queue, swapLocation}); return toSwap.size(); } @@ -376,13 +379,13 @@ public class FileSystemSwapManager implements FlowFileSwapManager { final int swapEncodingVersion = in.readInt(); if (swapEncodingVersion > SWAP_ENCODING_VERSION) { throw new IOException("Cannot swap FlowFiles in from SwapFile because the encoding version is " - + swapEncodingVersion + ", which is too new (expecting " + SWAP_ENCODING_VERSION + " or less)"); + + swapEncodingVersion + ", which is too new (expecting " + SWAP_ENCODING_VERSION + " or less)"); } final String connectionId = in.readUTF(); // Connection ID if (!connectionId.equals(queue.getIdentifier())) { - throw new IllegalArgumentException("Cannot deserialize FlowFiles from Swap File at location " + swapLocation + - " because those FlowFiles belong to Connection with ID " + connectionId + " and an attempt was made to swap them into a Connection with ID " + queue.getIdentifier()); + throw new IllegalArgumentException("Cannot deserialize FlowFiles from Swap File at location " + swapLocation + + " because those FlowFiles belong to Connection with ID " + connectionId + " and an attempt was made to swap them into a Connection with ID " + queue.getIdentifier()); } int numRecords = 0; @@ -396,8 +399,8 @@ public class FileSystemSwapManager implements FlowFileSwapManager { } } catch (final EOFException eof) { final QueueSize queueSize = new QueueSize(numRecords, contentSize); - final SwapSummary summary = new StandardSwapSummary(queueSize, maxRecordId, Collections.<ResourceClaim> emptyList()); - final SwapContents partialContents = new StandardSwapContents(summary, Collections.<FlowFileRecord> emptyList()); + final SwapSummary summary = new StandardSwapSummary(queueSize, maxRecordId, Collections.<ResourceClaim>emptyList()); + final SwapContents partialContents = new StandardSwapContents(summary, Collections.<FlowFileRecord>emptyList()); throw new IncompleteSwapFileException(swapLocation, partialContents); } @@ -406,7 +409,7 @@ public class FileSystemSwapManager implements FlowFileSwapManager { } private static SwapContents deserializeFlowFiles(final DataInputStream in, final QueueSize queueSize, final Long maxRecordId, - final int serializationVersion, final boolean incrementContentClaims, final ResourceClaimManager claimManager, final String location) throws IOException { + final int serializationVersion, final boolean incrementContentClaims, final ResourceClaimManager claimManager, final String location) throws IOException { final List<FlowFileRecord> flowFiles = new ArrayList<>(queueSize.getObjectCount()); final List<ResourceClaim> resourceClaims = new ArrayList<>(queueSize.getObjectCount()); Long maxId = maxRecordId; @@ -432,7 +435,7 @@ public class FileSystemSwapManager implements FlowFileSwapManager { if (serializationVersion > 1) { // Lineage information was added in version 2 - if(serializationVersion < 10){ + if (serializationVersion < 10) { final int numLineageIdentifiers = in.readInt(); for (int lineageIdIdx = 0; lineageIdIdx < numLineageIdentifiers; lineageIdIdx++) { in.readUTF(); //skip each identifier @@ -590,7 +593,6 @@ public class FileSystemSwapManager implements FlowFileSwapManager { } } - private void error(final String error) { logger.error(error); if (eventReporter != null) { @@ -605,9 +607,8 @@ public class FileSystemSwapManager implements FlowFileSwapManager { } } - - private static class SwapFileComparator implements Comparator<String> { + @Override public int compare(final String o1, final String o2) { if (o1 == o2) {
