public <T> PeerTaskExecutorResult<T> executeAgainstPeer(
      final PeerTask<T> peerTask, final EthPeer peer) {
    String taskClassName = peerTask.getClass().getSimpleName();
    AtomicInteger inflightRequestCountForThisTaskClass =
        inflightRequestCountByClassName.computeIfAbsent(
            taskClassName,
            (k) -> {
              AtomicInteger inflightRequests = new AtomicInteger(0);
              inflightRequestGauge.labels(inflightRequests::get, taskClassName);
              return inflightRequests;
            });
    MessageData requestMessageData = peerTask.getRequestMessage();
    SubProtocol peerTaskSubProtocol = peerTask.getSubProtocol();
    PeerTaskExecutorResult<T> executorResult;
    int retriesRemaining = peerTask.getRetriesWithSamePeer();
    do {
      try {
        T result;
        try (final OperationTimer.TimingContext ignored =
            requestTimer.labels(taskClassName).startTimer()) {
          inflightRequestCountForThisTaskClass.incrementAndGet();

          MessageData responseMessageData =
              requestSender.sendRequest(peerTaskSubProtocol, requestMessageData, peer);

          if (responseMessageData == null) {
            throw new InvalidPeerTaskResponseException();
          }

          result = peerTask.processResponse(responseMessageData);
        } finally {
          inflightRequestCountForThisTaskClass.decrementAndGet();
        }

        PeerTaskValidationResponse validationResponse = peerTask.validateResult(result);
        if (validationResponse == PeerTaskValidationResponse.RESULTS_VALID_AND_GOOD) {
          peer.recordUsefulResponse();
          executorResult =
              new PeerTaskExecutorResult<>(
                  Optional.ofNullable(result),
                  PeerTaskExecutorResponseCode.SUCCESS,
                  Optional.of(peer));
          peerTask.postProcessResult(executorResult);
        } else {
          LOG.debug(
              "Invalid response found for {} from peer {}", taskClassName, peer.getLoggableId());
          validationResponse.getDisconnectReason().ifPresent(peer::disconnect);
          executorResult =
              new PeerTaskExecutorResult<>(
                  Optional.ofNullable(result),
                  PeerTaskExecutorResponseCode.INVALID_RESPONSE,
                  Optional.of(peer));
        }

      } catch (PeerNotConnected e) {
        executorResult =
            new PeerTaskExecutorResult<>(
                Optional.empty(),
                PeerTaskExecutorResponseCode.PEER_DISCONNECTED,
                Optional.of(peer));

      } catch (InterruptedException | TimeoutException e) {
        peer.recordRequestTimeout(peerTaskSubProtocol.getName(), requestMessageData.getCode());
        timeoutCounter.labels(taskClassName).inc();
        executorResult =
            new PeerTaskExecutorResult<>(
                Optional.empty(), PeerTaskExecutorResponseCode.TIMEOUT, Optional.of(peer));

      } catch (InvalidPeerTaskResponseException e) {
        peer.recordUselessResponse(e.getMessage());
        invalidResponseCounter.labels(taskClassName).inc();
        LOG.debug(
            "Invalid response found for {} from peer {}", taskClassName, peer.getLoggableId(), e);
        executorResult =
            new PeerTaskExecutorResult<>(
                Optional.empty(), PeerTaskExecutorResponseCode.INVALID_RESPONSE, Optional.of(peer));

      } catch (Exception e) {
        internalExceptionCounter.labels(taskClassName).inc();
        LOG.error("Server error found for {} from peer {}", taskClassName, peer.getLoggableId(), e);
        executorResult =
            new PeerTaskExecutorResult<>(
                Optional.empty(),
                PeerTaskExecutorResponseCode.INTERNAL_SERVER_ERROR,
                Optional.of(peer));
      }
    } while (retriesRemaining-- > 0
        && executorResult.responseCode() != PeerTaskExecutorResponseCode.SUCCESS
        && executorResult.responseCode() != PeerTaskExecutorResponseCode.PEER_DISCONNECTED
        && sleepBetweenRetries());

    return executorResult;
  }
Manages the execution of PeerTasks, respecting their PeerTaskRetryBehavior
ethereum/eth/src/main/java/org/hyperledger/besu/ethereum/eth/manager/peertask/PeerTaskExecutor.java