@Test public void testFetchRemotePeerStatuses() throws IOException { final Set<PeerStatus> peerStatuses = new HashSet<>(); final PeerDescription bootstrapNode = new PeerDescription("Node1", 1111, true); final PeerDescription node2 = new PeerDescription("Node2", 2222, true); final PeerStatus bootstrapNodeStatus = new PeerStatus(bootstrapNode, 10, true); final PeerStatus node2Status = new PeerStatus(node2, 10, true); peerStatuses.add(bootstrapNodeStatus); peerStatuses.add(node2Status); final PeerStatusProvider peerStatusProvider = Mockito.mock(PeerStatusProvider.class); final PeerSelector peerSelector = new PeerSelector(peerStatusProvider, null); final UnitTestSystemTime systemTime = new UnitTestSystemTime(); peerSelector.setSystemTime(systemTime); doReturn(bootstrapNode).when(peerStatusProvider).getBootstrapPeerDescription(); doAnswer(invocation -> { final PeerDescription peerFetchStatusesFrom = invocation.getArgumentAt(0, PeerDescription.class); if (peerStatuses.stream().filter(ps -> ps.getPeerDescription().equals(peerFetchStatusesFrom)).collect(Collectors.toSet()).size() > 0) { return peerStatuses; } throw new IOException("Connection refused. " + peerFetchStatusesFrom + " is not running."); }).when(peerStatusProvider).fetchRemotePeerStatuses(any(PeerDescription.class)); peerSelector.refreshPeers(); PeerStatus peerStatus = peerSelector.getNextPeerStatus(TransferDirection.RECEIVE); assertNotNull(peerStatus); peerStatuses.remove(bootstrapNodeStatus); systemTime.offset += TimeUnit.MILLISECONDS.convert(1, TimeUnit.MINUTES) + 1; peerSelector.refreshPeers(); peerStatus = peerSelector.getNextPeerStatus(TransferDirection.RECEIVE); assertNotNull(peerStatus); assertEquals("Node2 should be returned since node 2 is the only available node.", node2, peerStatus.getPeerDescription()); systemTime.offset += TimeUnit.MILLISECONDS.convert(1, TimeUnit.MINUTES) + 1; peerSelector.refreshPeers(); peerStatus = peerSelector.getNextPeerStatus(TransferDirection.RECEIVE); assertNotNull(peerStatus); assertEquals("Node2 should be returned since node 2 is the only available node.", node2, peerStatus.getPeerDescription()); peerStatuses.remove(node2Status); systemTime.offset += TimeUnit.MILLISECONDS.convert(1, TimeUnit.MINUTES) + 1; peerSelector.refreshPeers(); peerStatus = peerSelector.getNextPeerStatus(TransferDirection.RECEIVE); assertNull("PeerSelector should return null as next peer status, since there's no available peer", peerStatus); peerStatuses.add(bootstrapNodeStatus); systemTime.offset += TimeUnit.MILLISECONDS.convert(1, TimeUnit.MINUTES) + 1; peerSelector.refreshPeers(); peerStatus = peerSelector.getNextPeerStatus(TransferDirection.RECEIVE); assertEquals("Node1 should be returned since node 1 is the only available node.", bootstrapNode, peerStatus.getPeerDescription()); }
private Set<PeerStatus> fetchRemotePeerStatuses() throws IOException { final Set<PeerDescription> peersToRequestClusterInfoFrom = new HashSet<>(); final Set<PeerStatus> lastFetched = lastFetchedQueryablePeers; if (lastFetched != null && !lastFetched.isEmpty()) { lastFetched.stream().map(peer -> peer.getPeerDescription()) .forEach(desc -> peersToRequestClusterInfoFrom.add(desc)); } peersToRequestClusterInfoFrom.add(peerStatusProvider.getBootstrapPeerDescription()); logger.debug("Fetching remote peer statuses from: {}", peersToRequestClusterInfoFrom); Exception lastFailure = null; for (final PeerDescription peerDescription : peersToRequestClusterInfoFrom) { try { final Set<PeerStatus> statuses = peerStatusProvider.fetchRemotePeerStatuses(peerDescription); lastFetchedQueryablePeers = statuses.stream() .filter(p -> p.isQueryForPeers()) .collect(Collectors.toSet()); return statuses; } catch (final Exception e) { logger.warn("Could not communicate with {}:{} to determine which nodes exist in the remote NiFi cluster, due to {}", peerDescription.getHostname(), peerDescription.getPort(), e.toString()); lastFailure = e; } } final IOException ioe = new IOException("Unable to communicate with remote NiFi cluster in order to determine which nodes exist in the remote cluster"); if (lastFailure != null) { ioe.addSuppressed(lastFailure); } throw ioe; }
PeerSelector { private Set<PeerStatus> fetchRemotePeerStatuses() throws IOException { final Set<PeerDescription> peersToRequestClusterInfoFrom = new HashSet<>(); final Set<PeerStatus> lastFetched = lastFetchedQueryablePeers; if (lastFetched != null && !lastFetched.isEmpty()) { lastFetched.stream().map(peer -> peer.getPeerDescription()) .forEach(desc -> peersToRequestClusterInfoFrom.add(desc)); } peersToRequestClusterInfoFrom.add(peerStatusProvider.getBootstrapPeerDescription()); logger.debug("Fetching remote peer statuses from: {}", peersToRequestClusterInfoFrom); Exception lastFailure = null; for (final PeerDescription peerDescription : peersToRequestClusterInfoFrom) { try { final Set<PeerStatus> statuses = peerStatusProvider.fetchRemotePeerStatuses(peerDescription); lastFetchedQueryablePeers = statuses.stream() .filter(p -> p.isQueryForPeers()) .collect(Collectors.toSet()); return statuses; } catch (final Exception e) { logger.warn("Could not communicate with {}:{} to determine which nodes exist in the remote NiFi cluster, due to {}", peerDescription.getHostname(), peerDescription.getPort(), e.toString()); lastFailure = e; } } final IOException ioe = new IOException("Unable to communicate with remote NiFi cluster in order to determine which nodes exist in the remote cluster"); if (lastFailure != null) { ioe.addSuppressed(lastFailure); } throw ioe; } }
PeerSelector { private Set<PeerStatus> fetchRemotePeerStatuses() throws IOException { final Set<PeerDescription> peersToRequestClusterInfoFrom = new HashSet<>(); final Set<PeerStatus> lastFetched = lastFetchedQueryablePeers; if (lastFetched != null && !lastFetched.isEmpty()) { lastFetched.stream().map(peer -> peer.getPeerDescription()) .forEach(desc -> peersToRequestClusterInfoFrom.add(desc)); } peersToRequestClusterInfoFrom.add(peerStatusProvider.getBootstrapPeerDescription()); logger.debug("Fetching remote peer statuses from: {}", peersToRequestClusterInfoFrom); Exception lastFailure = null; for (final PeerDescription peerDescription : peersToRequestClusterInfoFrom) { try { final Set<PeerStatus> statuses = peerStatusProvider.fetchRemotePeerStatuses(peerDescription); lastFetchedQueryablePeers = statuses.stream() .filter(p -> p.isQueryForPeers()) .collect(Collectors.toSet()); return statuses; } catch (final Exception e) { logger.warn("Could not communicate with {}:{} to determine which nodes exist in the remote NiFi cluster, due to {}", peerDescription.getHostname(), peerDescription.getPort(), e.toString()); lastFailure = e; } } final IOException ioe = new IOException("Unable to communicate with remote NiFi cluster in order to determine which nodes exist in the remote cluster"); if (lastFailure != null) { ioe.addSuppressed(lastFailure); } throw ioe; } PeerSelector(final PeerStatusProvider peerStatusProvider, final File persistenceFile); }
PeerSelector { private Set<PeerStatus> fetchRemotePeerStatuses() throws IOException { final Set<PeerDescription> peersToRequestClusterInfoFrom = new HashSet<>(); final Set<PeerStatus> lastFetched = lastFetchedQueryablePeers; if (lastFetched != null && !lastFetched.isEmpty()) { lastFetched.stream().map(peer -> peer.getPeerDescription()) .forEach(desc -> peersToRequestClusterInfoFrom.add(desc)); } peersToRequestClusterInfoFrom.add(peerStatusProvider.getBootstrapPeerDescription()); logger.debug("Fetching remote peer statuses from: {}", peersToRequestClusterInfoFrom); Exception lastFailure = null; for (final PeerDescription peerDescription : peersToRequestClusterInfoFrom) { try { final Set<PeerStatus> statuses = peerStatusProvider.fetchRemotePeerStatuses(peerDescription); lastFetchedQueryablePeers = statuses.stream() .filter(p -> p.isQueryForPeers()) .collect(Collectors.toSet()); return statuses; } catch (final Exception e) { logger.warn("Could not communicate with {}:{} to determine which nodes exist in the remote NiFi cluster, due to {}", peerDescription.getHostname(), peerDescription.getPort(), e.toString()); lastFailure = e; } } final IOException ioe = new IOException("Unable to communicate with remote NiFi cluster in order to determine which nodes exist in the remote cluster"); if (lastFailure != null) { ioe.addSuppressed(lastFailure); } throw ioe; } PeerSelector(final PeerStatusProvider peerStatusProvider, final File persistenceFile); void penalize(final Peer peer, final long penalizationMillis); void penalize(final PeerDescription peerDescription, final long penalizationMillis); boolean isPenalized(final PeerStatus peerStatus); void clear(); PeerStatus getNextPeerStatus(final TransferDirection direction); void refreshPeers(); void setEventReporter(EventReporter eventReporter); }
PeerSelector { private Set<PeerStatus> fetchRemotePeerStatuses() throws IOException { final Set<PeerDescription> peersToRequestClusterInfoFrom = new HashSet<>(); final Set<PeerStatus> lastFetched = lastFetchedQueryablePeers; if (lastFetched != null && !lastFetched.isEmpty()) { lastFetched.stream().map(peer -> peer.getPeerDescription()) .forEach(desc -> peersToRequestClusterInfoFrom.add(desc)); } peersToRequestClusterInfoFrom.add(peerStatusProvider.getBootstrapPeerDescription()); logger.debug("Fetching remote peer statuses from: {}", peersToRequestClusterInfoFrom); Exception lastFailure = null; for (final PeerDescription peerDescription : peersToRequestClusterInfoFrom) { try { final Set<PeerStatus> statuses = peerStatusProvider.fetchRemotePeerStatuses(peerDescription); lastFetchedQueryablePeers = statuses.stream() .filter(p -> p.isQueryForPeers()) .collect(Collectors.toSet()); return statuses; } catch (final Exception e) { logger.warn("Could not communicate with {}:{} to determine which nodes exist in the remote NiFi cluster, due to {}", peerDescription.getHostname(), peerDescription.getPort(), e.toString()); lastFailure = e; } } final IOException ioe = new IOException("Unable to communicate with remote NiFi cluster in order to determine which nodes exist in the remote cluster"); if (lastFailure != null) { ioe.addSuppressed(lastFailure); } throw ioe; } PeerSelector(final PeerStatusProvider peerStatusProvider, final File persistenceFile); void penalize(final Peer peer, final long penalizationMillis); void penalize(final PeerDescription peerDescription, final long penalizationMillis); boolean isPenalized(final PeerStatus peerStatus); void clear(); PeerStatus getNextPeerStatus(final TransferDirection direction); void refreshPeers(); void setEventReporter(EventReporter eventReporter);  }