docs: 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;
        }
      }
    }