private FullResponseHolder submitRequest(
      String taskId,
      @Nullable String mediaType, // nullable if content is empty
      HttpMethod method,
      String encodedPathSuffix,
      @Nullable String encodedQueryString,
      byte[] content,
      boolean retry
  ) throws IOException, ChannelException, NoTaskLocationException
  {
    final RetryPolicy retryPolicy = retryPolicyFactory.makeRetryPolicy();

    while (true) {
      String path = StringUtils.format("%s/%s/%s", BASE_PATH, StringUtils.urlEncode(taskId), encodedPathSuffix);

      Optional<TaskStatus> status = taskInfoProvider.getTaskStatus(taskId);
      if (!status.isPresent() || !status.get().isRunnable()) {
        throw new TaskNotRunnableException(
            StringUtils.format(
                "Aborting request because task [%s] is not runnable",
                taskId
            )
        );
      }

      final TaskLocation location = taskInfoProvider.getTaskLocation(taskId);
      if (location.equals(TaskLocation.unknown())) {
        throw new NoTaskLocationException(StringUtils.format("No TaskLocation available for task [%s]", taskId));
      }

      final Request request = createRequest(
          taskId,
          location,
          path,
          encodedQueryString,
          method,
          mediaType,
          content
      );

      FullResponseHolder response = null;
      try {
        // Netty throws some annoying exceptions if a connection can't be opened, which happens relatively frequently
        // for tasks that happen to still be starting up, so test the connection first to keep the logs clean.
        checkConnection(request.getUrl().getHost(), request.getUrl().getPort());

        response = submitRequest(request);

        int responseCode = response.getStatus().getCode();
        if (responseCode / 100 == 2) {
          return response;
        } else if (responseCode == 400) { // don't bother retrying if it's a bad request
          throw new IAE("Received 400 Bad Request with body: %s", response.getContent());
        } else {
          throw new IOE("Received status [%d] and content [%s]", responseCode, response.getContent());
        }
      }
      catch (IOException | ChannelException e) {

        // Since workers are free to move tasks around to different ports, there is a chance that a task may have been
        // moved but our view of its location has not been updated yet from ZK. To detect this case, we send a header
        // identifying our expected recipient in the request; if this doesn't correspond to the worker we messaged, the
        // worker will return an HTTP 404 with its ID in the response header. If we get a mismatching task ID, then
        // we will wait for a short period then retry the request indefinitely, expecting the task's location to
        // eventually be updated.

        final Duration delay;
        if (response != null && response.getStatus().equals(HttpResponseStatus.NOT_FOUND)) {
          String headerId = StringUtils.urlDecode(
              response.getResponse().headers().get(ChatHandlerResource.TASK_ID_HEADER)
          );
          if (headerId != null && !headerId.equals(taskId)) {
            log.warn(
                "Expected worker to have taskId [%s] but has taskId [%s], will retry in [%d]s",
                taskId, headerId, TASK_MISMATCH_RETRY_DELAY_SECONDS
            );
            delay = Duration.standardSeconds(TASK_MISMATCH_RETRY_DELAY_SECONDS);
          } else {
            delay = retryPolicy.getAndIncrementRetryDelay();
          }
        } else {
          delay = retryPolicy.getAndIncrementRetryDelay();
        }
        final String urlForLog = request.getUrl().toString();
        if (!retry) {
          // if retry=false, we probably aren't too concerned if the operation doesn't succeed (i.e. the request was
          // for informational purposes only) so don't log a scary stack trace
          log.info("submitRequest failed for [%s], with message [%s]", urlForLog, e.getMessage());
          throw e;
        } else if (delay == null) {
          log.warn(e, "Retries exhausted for [%s], last exception:", urlForLog);
          throw e;
        } else {
          try {
            final long sleepTime = delay.getMillis();
            log.debug(
                "Bad response HTTP [%s] from [%s]; will try again in [%s] (body/exception: [%s])",
                (response != null ? response.getStatus().getCode() : "no response"),
                urlForLog,
                new Duration(sleepTime).toString(),
                (response != null ? response.getContent() : e.getMessage())
            );
            Thread.sleep(sleepTime);
          }
          catch (InterruptedException e2) {
            Thread.currentThread().interrupt();
            e.addSuppressed(e2);
            throw new RuntimeException(e);
          }
        }
      }
      catch (NoTaskLocationException e) {
        log.info("No TaskLocation available for task [%s], this task may not have been assigned to a worker yet or "
                 + "may have already completed", taskId);
        throw e;
      }
      catch (Exception e) {
        log.warn(e, "Exception while sending request");
        throw e;
      }
    }
  }
private FullResponseHolder submitRequest ( String taskId , @ Nullable String mediaType , HttpMethod method , String encodedPathSuffix , @ Nullable String encodedQueryString , byte [ ] content , boolean retry ) throws IOException , ChannelException , NoTaskLocationException { final RetryPolicy retryPolicy = retryPolicyFactory . makeRetryPolicy ( ) ; while ( true ) { String path = StringUtils . format ( "%s/%s/%s" , BASE_PATH , StringUtils . urlEncode ( taskId ) , encodedPathSuffix ) ; Optional < TaskStatus > status = taskInfoProvider . getTaskStatus ( taskId ) ; if ( ! status . isPresent ( ) || ! status . get ( ) . isRunnable ( ) ) { throw new TaskNotRunnableException ( StringUtils . format ( "Aborting request because task [%s] is not runnable" , taskId ) ) ; } final TaskLocation location = taskInfoProvider . getTaskLocation ( taskId ) ; if ( location . equals ( TaskLocation . unknown ( ) ) ) { throw new NoTaskLocationException ( StringUtils . format ( "No TaskLocation available for task [%s]" , taskId ) ) ; } final Request request = createRequest ( taskId , location , path , encodedQueryString , method , mediaType , content ) ; FullResponseHolder response = null ; try { checkConnection ( request . getUrl ( ) . getHost ( ) , request . getUrl ( ) . getPort ( ) ) ; response = submitRequest ( request ) ; int responseCode = response . getStatus ( ) . getCode ( ) ; if ( responseCode / 100 == 2 ) { return response ; } else if ( responseCode == 400 ) { throw new IAE ( "Received 400 Bad Request with body: %s" , response . getContent ( ) ) ; } else { throw new IOE ( "Received status [%d] and content [%s]" , responseCode , response . getContent ( ) ) ; } } catch ( IOException | ChannelException e ) { final Duration delay ; if ( response != null && response . getStatus ( ) . equals ( HttpResponseStatus . NOT_FOUND ) ) { String headerId = StringUtils . urlDecode ( response . getResponse ( ) . headers ( ) . get ( ChatHandlerResource . TASK_ID_HEADER ) ) ; if ( headerId != null && ! headerId . equals ( taskId ) ) { log . warn ( "Expected worker to have taskId [%s] but has taskId [%s], will retry in [%d]s" , taskId , headerId , TASK_MISMATCH_RETRY_DELAY_SECONDS ) ; delay = Duration . standardSeconds ( TASK_MISMATCH_RETRY_DELAY_SECONDS ) ; } else { delay = retryPolicy . getAndIncrementRetryDelay ( ) ; } } else { delay = retryPolicy . getAndIncrementRetryDelay ( ) ; } final String urlForLog = request . getUrl ( ) . toString ( ) ; if ( ! retry ) { log . info ( "submitRequest failed for [%s], with message [%s]" , urlForLog , e . getMessage ( ) ) ; throw e ; } else if ( delay == null ) { log . warn ( e , "Retries exhausted for [%s], last exception:" , urlForLog ) ; throw e ; } else { try { final long sleepTime = delay . getMillis ( ) ; log . debug ( "Bad response HTTP [%s] from [%s]; will try again in [%s] (body/exception: [%s])" , ( response != null ? response . getStatus ( ) . getCode ( ) : "no response" ) , urlForLog , new Duration ( sleepTime ) . toString ( ) , ( response != null ? response . getContent ( ) : e . getMessage ( ) ) ) ; Thread . sleep ( sleepTime ) ; } catch ( InterruptedException e2 ) { Thread . currentThread ( ) . interrupt ( ) ; e . addSuppressed ( e2 ) ; throw new RuntimeException ( e ) ; } } } catch ( NoTaskLocationException e ) { log . info ( "No TaskLocation available for task [%s], this task may not have been assigned to a worker yet or " + "may have already completed" , taskId ) ; throw e ; } catch ( Exception e ) { log . warn ( e , "Exception while sending request" ) ; throw e ; } } }
Sends an HTTP request to the task of the specified {@code taskId} and returns a response if it succeeded.