class 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 the request is mutable, ensure that all nodes are connected.
        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);
    }

    public  ThreadPoolRequestReplicator(final int numThreads, final Client client, final ClusterCoordinator clusterCoordinator,
                                       final RequestCompletionCallback callback, final EventReporter eventReporter, final NiFiProperties nifiProperties);
    public  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);

}