diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthPeers.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthPeers.java index 1b413251be..47ceba6598 100644 --- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthPeers.java +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/EthPeers.java @@ -17,7 +17,6 @@ package org.hyperledger.besu.ethereum.eth.manager; import org.hyperledger.besu.ethereum.core.BlockHeader; import org.hyperledger.besu.ethereum.eth.SnapProtocol; import org.hyperledger.besu.ethereum.eth.manager.EthPeer.DisconnectCallback; -import org.hyperledger.besu.ethereum.eth.manager.exceptions.NoAvailablePeersException; import org.hyperledger.besu.ethereum.eth.manager.peertask.PeerSelector; import org.hyperledger.besu.ethereum.eth.peervalidation.PeerValidator; import org.hyperledger.besu.ethereum.eth.sync.ChainHeadTracker; @@ -470,8 +469,8 @@ public class EthPeers implements PeerSelector { // Part of the PeerSelector interface, to be split apart later @Override - public EthPeer getPeer(final Predicate filter) { - return streamBestPeers().filter(filter).findFirst().orElseThrow(NoAvailablePeersException::new); + public Optional getPeer(final Predicate filter) { + return bestPeerMatchingCriteria(filter); } // Part of the PeerSelector interface, to be split apart later diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerSelector.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerSelector.java index 0801f9f00e..8f7ab33e42 100644 --- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerSelector.java +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerSelector.java @@ -29,7 +29,7 @@ public interface PeerSelector { * @param filter a Predicate\ matching desirable peers * @return a peer matching the supplied conditions */ - EthPeer getPeer(final Predicate filter); + Optional getPeer(final Predicate filter); /** * Attempts to get the EthPeer identified by peerId diff --git a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutor.java b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutor.java index 7e38d8cfa7..1734c1e876 100644 --- a/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutor.java +++ b/ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutor.java @@ -15,7 +15,6 @@ package org.hyperledger.besu.ethereum.eth.manager.peertask; import org.hyperledger.besu.ethereum.eth.manager.EthPeer; -import org.hyperledger.besu.ethereum.eth.manager.exceptions.NoAvailablePeersException; import org.hyperledger.besu.ethereum.p2p.rlpx.connections.PeerConnection; import org.hyperledger.besu.ethereum.p2p.rlpx.wire.MessageData; import org.hyperledger.besu.metrics.BesuMetricCategory; @@ -62,20 +61,19 @@ public class PeerTaskExecutor { : NO_RETRIES; final Collection usedEthPeers = new HashSet<>(); do { - EthPeer peer; - try { - peer = - peerSelector.getPeer( - (candidatePeer) -> - peerTask.getPeerRequirementFilter().test(candidatePeer) - && !usedEthPeers.contains(candidatePeer)); - usedEthPeers.add(peer); - executorResult = executeAgainstPeer(peerTask, peer); - } catch (NoAvailablePeersException e) { + Optional peer = + peerSelector.getPeer( + (candidatePeer) -> + peerTask.getPeerRequirementFilter().test(candidatePeer) + && !usedEthPeers.contains(candidatePeer)); + if (peer.isEmpty()) { executorResult = new PeerTaskExecutorResult<>( Optional.empty(), PeerTaskExecutorResponseCode.NO_PEER_AVAILABLE); + continue; } + usedEthPeers.add(peer.get()); + executorResult = executeAgainstPeer(peerTask, peer.get()); } while (retriesRemaining-- > 0 && executorResult.responseCode() != PeerTaskExecutorResponseCode.SUCCESS); diff --git a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutorTest.java b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutorTest.java index 55878d144e..6dfd8d0e20 100644 --- a/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutorTest.java +++ b/ethereum/eth/src/test/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutorTest.java @@ -22,6 +22,7 @@ import org.hyperledger.besu.metrics.noop.NoOpMetricsSystem; import java.util.Collections; import java.util.List; +import java.util.Optional; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeoutException; import java.util.function.Predicate; @@ -49,8 +50,7 @@ public class PeerTaskExecutorTest { @BeforeEach public void beforeTest() { mockCloser = MockitoAnnotations.openMocks(this); - peerTaskExecutor = - new PeerTaskExecutor(peerSelector, requestSender, new NoOpMetricsSystem()); + peerTaskExecutor = new PeerTaskExecutor(peerSelector, requestSender, new NoOpMetricsSystem()); } @AfterEach @@ -201,7 +201,8 @@ public class PeerTaskExecutorTest { InvalidPeerTaskResponseException { Object responseObject = new Object(); - Mockito.when(peerSelector.getPeer(Mockito.any(Predicate.class))).thenReturn(ethPeer); + Mockito.when(peerSelector.getPeer(Mockito.any(Predicate.class))) + .thenReturn(Optional.of(ethPeer)); Mockito.when(peerTask.getRequestMessage()).thenReturn(requestMessageData); Mockito.when(peerTask.getPeerTaskRetryBehaviors()).thenReturn(Collections.emptyList()); @@ -234,8 +235,8 @@ public class PeerTaskExecutorTest { EthPeer peer2 = Mockito.mock(EthPeer.class); Mockito.when(peerSelector.getPeer(Mockito.any(Predicate.class))) - .thenReturn(ethPeer) - .thenReturn(peer2); + .thenReturn(Optional.of(ethPeer)) + .thenReturn(Optional.of(peer2)); Mockito.when(peerTask.getRequestMessage()).thenReturn(requestMessageData); Mockito.when(peerTask.getPeerTaskRetryBehaviors())