@Test(timeout = 15000) public void testCompleteOnError() { withReplicator(replicator -> { final Set<NodeIdentifier> nodeIds = new HashSet<>(); final NodeIdentifier id1 = new NodeIdentifier("1", "localhost", 8100, "localhost", 8101, "localhost", 8102, 8103, false); final NodeIdentifier id2 = new NodeIdentifier("2", "localhost", 8200, "localhost", 8201, "localhost", 8202, 8203, false); final NodeIdentifier id3 = new NodeIdentifier("3", "localhost", 8300, "localhost", 8301, "localhost", 8302, 8303, false); final NodeIdentifier id4 = new NodeIdentifier("4", "localhost", 8400, "localhost", 8401, "localhost", 8402, 8403, false); nodeIds.add(id1); nodeIds.add(id2); nodeIds.add(id3); nodeIds.add(id4); final URI uri = new URI("http: final Entity entity = new ProcessorEntity(); final Authentication authentication = new NiFiAuthenticationToken(new NiFiUserDetails(StandardNiFiUser.ANONYMOUS)); SecurityContextHolder.getContext().setAuthentication(authentication); 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")); }
@Override public AsyncClusterResponse replicate(String method, URI uri, Object entity, Map<String, String> headers) { final Map<NodeConnectionState, List<NodeIdentifier>> stateMap = clusterCoordinator.getConnectionStates(); final boolean mutable = isMutableRequest(method, uri.getPath()); if (mutable) { final List<NodeIdentifier> disconnected = stateMap.get(NodeConnectionState.DISCONNECTED); if (disconnected != null && !disconnected.isEmpty()) { if (disconnected.size() == 1) { throw new DisconnectedNodeMutableRequestException("Node " + disconnected.iterator().next() + " is currently disconnected"); } else { throw new DisconnectedNodeMutableRequestException(disconnected.size() + " Nodes are currently disconnected"); } } final List<NodeIdentifier> disconnecting = stateMap.get(NodeConnectionState.DISCONNECTING); if (disconnecting != null && !disconnecting.isEmpty()) { if (disconnecting.size() == 1) { throw new DisconnectedNodeMutableRequestException("Node " + disconnecting.iterator().next() + " is currently disconnecting"); } else { throw new DisconnectedNodeMutableRequestException(disconnecting.size() + " Nodes are currently disconnecting"); } } final List<NodeIdentifier> connecting = stateMap.get(NodeConnectionState.CONNECTING); if (connecting != null && !connecting.isEmpty()) { if (connecting.size() == 1) { throw new ConnectingNodeMutableRequestException("Node " + connecting.iterator().next() + " is currently connecting"); } else { throw new ConnectingNodeMutableRequestException(connecting.size() + " Nodes are currently connecting"); } } } final List<NodeIdentifier> nodeIds = stateMap.get(NodeConnectionState.CONNECTED); if (nodeIds == null || nodeIds.isEmpty()) { throw new NoConnectedNodesException(); } final Set<NodeIdentifier> nodeIdSet = new HashSet<>(nodeIds); return replicate(nodeIdSet, method, uri, entity, headers, true, true); }
ThreadPoolRequestReplicator implements RequestReplicator { @Override public AsyncClusterResponse replicate(String method, URI uri, Object entity, Map<String, String> headers) { final Map<NodeConnectionState, List<NodeIdentifier>> stateMap = clusterCoordinator.getConnectionStates(); final boolean mutable = isMutableRequest(method, uri.getPath()); if (mutable) { final List<NodeIdentifier> disconnected = stateMap.get(NodeConnectionState.DISCONNECTED); if (disconnected != null && !disconnected.isEmpty()) { if (disconnected.size() == 1) { throw new DisconnectedNodeMutableRequestException("Node " + disconnected.iterator().next() + " is currently disconnected"); } else { throw new DisconnectedNodeMutableRequestException(disconnected.size() + " Nodes are currently disconnected"); } } final List<NodeIdentifier> disconnecting = stateMap.get(NodeConnectionState.DISCONNECTING); if (disconnecting != null && !disconnecting.isEmpty()) { if (disconnecting.size() == 1) { throw new DisconnectedNodeMutableRequestException("Node " + disconnecting.iterator().next() + " is currently disconnecting"); } else { throw new DisconnectedNodeMutableRequestException(disconnecting.size() + " Nodes are currently disconnecting"); } } final List<NodeIdentifier> connecting = stateMap.get(NodeConnectionState.CONNECTING); if (connecting != null && !connecting.isEmpty()) { if (connecting.size() == 1) { throw new ConnectingNodeMutableRequestException("Node " + connecting.iterator().next() + " is currently connecting"); } else { throw new ConnectingNodeMutableRequestException(connecting.size() + " Nodes are currently connecting"); } } } final List<NodeIdentifier> nodeIds = stateMap.get(NodeConnectionState.CONNECTED); if (nodeIds == null || nodeIds.isEmpty()) { throw new NoConnectedNodesException(); } final Set<NodeIdentifier> nodeIdSet = new HashSet<>(nodeIds); return replicate(nodeIdSet, method, uri, entity, headers, true, true); } }
ThreadPoolRequestReplicator implements RequestReplicator { @Override public AsyncClusterResponse replicate(String method, URI uri, Object entity, Map<String, String> headers) { final Map<NodeConnectionState, List<NodeIdentifier>> stateMap = clusterCoordinator.getConnectionStates(); final boolean mutable = isMutableRequest(method, uri.getPath()); if (mutable) { final List<NodeIdentifier> disconnected = stateMap.get(NodeConnectionState.DISCONNECTED); if (disconnected != null && !disconnected.isEmpty()) { if (disconnected.size() == 1) { throw new DisconnectedNodeMutableRequestException("Node " + disconnected.iterator().next() + " is currently disconnected"); } else { throw new DisconnectedNodeMutableRequestException(disconnected.size() + " Nodes are currently disconnected"); } } final List<NodeIdentifier> disconnecting = stateMap.get(NodeConnectionState.DISCONNECTING); if (disconnecting != null && !disconnecting.isEmpty()) { if (disconnecting.size() == 1) { throw new DisconnectedNodeMutableRequestException("Node " + disconnecting.iterator().next() + " is currently disconnecting"); } else { throw new DisconnectedNodeMutableRequestException(disconnecting.size() + " Nodes are currently disconnecting"); } } final List<NodeIdentifier> connecting = stateMap.get(NodeConnectionState.CONNECTING); if (connecting != null && !connecting.isEmpty()) { if (connecting.size() == 1) { throw new ConnectingNodeMutableRequestException("Node " + connecting.iterator().next() + " is currently connecting"); } else { throw new ConnectingNodeMutableRequestException(connecting.size() + " Nodes are currently connecting"); } } } final List<NodeIdentifier> nodeIds = stateMap.get(NodeConnectionState.CONNECTED); if (nodeIds == null || nodeIds.isEmpty()) { throw new NoConnectedNodesException(); } final Set<NodeIdentifier> nodeIdSet = new HashSet<>(nodeIds); return replicate(nodeIdSet, method, uri, entity, headers, true, true); } ThreadPoolRequestReplicator(final int numThreads, final Client client, final ClusterCoordinator clusterCoordinator,
                                       final RequestCompletionCallback callback, final EventReporter eventReporter, final NiFiProperties nifiProperties);  ThreadPoolRequestReplicator(final int numThreads, final Client client, final ClusterCoordinator clusterCoordinator,
                                       final String connectionTimeout, final String readTimeout, final RequestCompletionCallback callback,
                                       final EventReporter eventReporter, final NiFiProperties nifiProperties); }
ThreadPoolRequestReplicator implements RequestReplicator { @Override public AsyncClusterResponse replicate(String method, URI uri, Object entity, Map<String, String> headers) { final Map<NodeConnectionState, List<NodeIdentifier>> stateMap = clusterCoordinator.getConnectionStates(); final boolean mutable = isMutableRequest(method, uri.getPath()); if (mutable) { final List<NodeIdentifier> disconnected = stateMap.get(NodeConnectionState.DISCONNECTED); if (disconnected != null && !disconnected.isEmpty()) { if (disconnected.size() == 1) { throw new DisconnectedNodeMutableRequestException("Node " + disconnected.iterator().next() + " is currently disconnected"); } else { throw new DisconnectedNodeMutableRequestException(disconnected.size() + " Nodes are currently disconnected"); } } final List<NodeIdentifier> disconnecting = stateMap.get(NodeConnectionState.DISCONNECTING); if (disconnecting != null && !disconnecting.isEmpty()) { if (disconnecting.size() == 1) { throw new DisconnectedNodeMutableRequestException("Node " + disconnecting.iterator().next() + " is currently disconnecting"); } else { throw new DisconnectedNodeMutableRequestException(disconnecting.size() + " Nodes are currently disconnecting"); } } final List<NodeIdentifier> connecting = stateMap.get(NodeConnectionState.CONNECTING); if (connecting != null && !connecting.isEmpty()) { if (connecting.size() == 1) { throw new ConnectingNodeMutableRequestException("Node " + connecting.iterator().next() + " is currently connecting"); } else { throw new ConnectingNodeMutableRequestException(connecting.size() + " Nodes are currently connecting"); } } } final List<NodeIdentifier> nodeIds = stateMap.get(NodeConnectionState.CONNECTED); if (nodeIds == null || nodeIds.isEmpty()) { throw new NoConnectedNodesException(); } final Set<NodeIdentifier> nodeIdSet = new HashSet<>(nodeIds); return replicate(nodeIdSet, method, uri, entity, headers, true, true); } ThreadPoolRequestReplicator(final int numThreads, final Client client, final ClusterCoordinator clusterCoordinator,
                                       final RequestCompletionCallback callback, final EventReporter eventReporter, final NiFiProperties nifiProperties);  ThreadPoolRequestReplicator(final int numThreads, final Client client, final ClusterCoordinator clusterCoordinator,
                                       final String connectionTimeout, final String readTimeout, final RequestCompletionCallback callback,
                                       final EventReporter eventReporter, final NiFiProperties nifiProperties); @Override void shutdown(); @Override AsyncClusterResponse replicate(String method, URI uri, Object entity, Map<String, String> headers); @Override AsyncClusterResponse replicate(Set<NodeIdentifier> nodeIds, String method, URI uri, Object entity, Map<String, String> headers,
                                          final boolean indicateReplicated, final boolean performVerification); @Override AsyncClusterResponse forwardToCoordinator(final NodeIdentifier coordinatorNodeId, final String method, final URI uri, final Object entity, final Map<String, String> headers); @Override AsyncClusterResponse getClusterResponse(final String identifier); }
ThreadPoolRequestReplicator implements RequestReplicator { @Override public AsyncClusterResponse replicate(String method, URI uri, Object entity, Map<String, String> headers) { final Map<NodeConnectionState, List<NodeIdentifier>> stateMap = clusterCoordinator.getConnectionStates(); final boolean mutable = isMutableRequest(method, uri.getPath()); if (mutable) { final List<NodeIdentifier> disconnected = stateMap.get(NodeConnectionState.DISCONNECTED); if (disconnected != null && !disconnected.isEmpty()) { if (disconnected.size() == 1) { throw new DisconnectedNodeMutableRequestException("Node " + disconnected.iterator().next() + " is currently disconnected"); } else { throw new DisconnectedNodeMutableRequestException(disconnected.size() + " Nodes are currently disconnected"); } } final List<NodeIdentifier> disconnecting = stateMap.get(NodeConnectionState.DISCONNECTING); if (disconnecting != null && !disconnecting.isEmpty()) { if (disconnecting.size() == 1) { throw new DisconnectedNodeMutableRequestException("Node " + disconnecting.iterator().next() + " is currently disconnecting"); } else { throw new DisconnectedNodeMutableRequestException(disconnecting.size() + " Nodes are currently disconnecting"); } } final List<NodeIdentifier> connecting = stateMap.get(NodeConnectionState.CONNECTING); if (connecting != null && !connecting.isEmpty()) { if (connecting.size() == 1) { throw new ConnectingNodeMutableRequestException("Node " + connecting.iterator().next() + " is currently connecting"); } else { throw new ConnectingNodeMutableRequestException(connecting.size() + " Nodes are currently connecting"); } } } final List<NodeIdentifier> nodeIds = stateMap.get(NodeConnectionState.CONNECTED); if (nodeIds == null || nodeIds.isEmpty()) { throw new NoConnectedNodesException(); } final Set<NodeIdentifier> nodeIdSet = new HashSet<>(nodeIds); return replicate(nodeIdSet, method, uri, entity, headers, true, true); } ThreadPoolRequestReplicator(final int numThreads, final Client client, final ClusterCoordinator clusterCoordinator,
                                       final RequestCompletionCallback callback, final EventReporter eventReporter, final NiFiProperties nifiProperties);  ThreadPoolRequestReplicator(final int numThreads, final Client client, final ClusterCoordinator clusterCoordinator,
                                       final String connectionTimeout, final String readTimeout, final RequestCompletionCallback callback,
                                       final EventReporter eventReporter, final NiFiProperties nifiProperties); @Override void shutdown(); @Override AsyncClusterResponse replicate(String method, URI uri, Object entity, Map<String, String> headers); @Override AsyncClusterResponse replicate(Set<NodeIdentifier> nodeIds, String method, URI uri, Object entity, Map<String, String> headers,
                                          final boolean indicateReplicated, final boolean performVerification); @Override AsyncClusterResponse forwardToCoordinator(final NodeIdentifier coordinatorNodeId, final String method, final URI uri, final Object entity, final Map<String, String> headers); @Override AsyncClusterResponse getClusterResponse(final String identifier);  }