import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
/** Leases and returns all tasks from the queue with the specified tag tld, in batches. */
public static List<TaskHandle> loadAllTasks(Queue queue, String tld) {
  ImmutableList.Builder<TaskHandle> allTasks = new ImmutableList.Builder<>();
  int numErrors = 0;
  long backOff = backOffMillis;
  while (true) {
    try {
      List<TaskHandle> tasks = queue.leaseTasks(LeaseOptions.Builder
          .withTag(tld)
          .leasePeriod(LEASE_PERIOD.getMillis(), TimeUnit.MILLISECONDS)
          .countLimit(BATCH_SIZE));
      allTasks.addAll(tasks);
      if (tasks.isEmpty()) {
        return allTasks.build();
      }
    } catch (TransientFailureException | DeadlineExceededException e) {
      if (++numErrors >= 3) {
        throw new RuntimeException("Error leasing tasks", e);
      }
      Uninterruptibles.sleepUninterruptibly(backOff, TimeUnit.MILLISECONDS);
      backOff *= 2;
    }
  }
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
public void export(BigQueryMetric metric) {
  try {
    String hostname = modulesService.getVersionHostname("backend", null);
    TaskOptions opts =
        withUrl(MetricsExportAction.PATH)
            .header("Host", hostname)
            .param("insertId", idGenerator.get());
    for (Entry<String, String> entry : metric.getBigQueryRowEncoding().entrySet()) {
      opts.param(entry.getKey(), entry.getValue());
    }
    opts.param("tableId", metric.getTableId());
    queue.add(opts);
  } catch (TransientFailureException e) {
    // Log and swallow. We may drop some metrics here but this should be rare.
    logger.info(e, e.getMessage());
  }
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
/** Deletes a list of tasks from the given queue using a retrier. */
private static void deleteTasksWithRetry(
    final List<DnsRefreshRequest> refreshRequests,
    final Queue queue,
    AsyncFlowMetrics asyncFlowMetrics,
    Retrier retrier,
    OperationResult result) {
  if (refreshRequests.isEmpty()) {
    return;
  }
  final List<TaskHandle> tasks =
      refreshRequests.stream().map(DnsRefreshRequest::task).collect(toImmutableList());
  retrier.callWithRetry(() -> queue.deleteTask(tasks), TransientFailureException.class);
  refreshRequests.forEach(
      r -> asyncFlowMetrics.recordAsyncFlowResult(DNS_REFRESH, result, r.requestedTime()));
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
/**
 * @param name
 * @param id
 */
public static void queueToIndex (String name, Long id) {
	Queue queue = QueueFactory.getDefaultQueue();

	TaskOptions options = TaskOptions.Builder.withMethod(Method.POST)
			.url(INDEX_SEARCH_URL).param(ENTITY_NAME_KEY, name)
			.param(ENTITY_ID_KEY, id.toString());

	int retry = RETRY_COUNT;
	do {
		try {
			queue.add(options);

			// success no need to retry
			retry = 0;
		} catch (TransientFailureException ex) {
			retry--;
		}
	} while (retry > 0);
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
@Override
public void enqueueTask(String taskName, Map<String, String[]> paramMap) {

  checkNotNull(taskName);

  final Queue queue = QueueFactory.getDefaultQueue();
  final TaskOptions options = TaskOptions.Builder.withUrl(String.format(PATH_ADMIN_TASK, taskName));

  for (Map.Entry<String, String[]> param : paramMap.entrySet()) {
    for (String value : param.getValue()) {
      options.param(param.getKey(), value);
    }
  }

  try {
    queue.add(options);
    LOGGER.info("Added admin task to queue {}", taskName);
  } catch (TransientFailureException tfe) {
    LOGGER.error("Run admin task fail {} ", tfe.getMessage());
  }

}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
/**
 * Transact with commit logs and retry with exponential backoff.
 *
 * <p>This method is broken out from {@link #transactNew(Work)} for testing purposes.
 */
@VisibleForTesting
<R> R transactCommitLoggedWork(CommitLoggedWork<R> work) {
  long baseRetryMillis = getBaseOfyRetryDuration().getMillis();
  for (long attempt = 0, sleepMillis = baseRetryMillis;
      true;
      attempt++, sleepMillis *= 2) {
    try {
      ofy().transactNew(work);
      return work.getResult();
    } catch (TransientFailureException
        | TimestampInversionException
        | DatastoreTimeoutException
        | DatastoreFailureException e) {
      // TransientFailureExceptions come from task queues and always mean nothing committed.
      // TimestampInversionExceptions are thrown by our code and are always retryable as well.
      // However, Datastore exceptions might get thrown even if the transaction succeeded.
      if ((e instanceof DatastoreTimeoutException || e instanceof DatastoreFailureException)
          && checkIfAlreadySucceeded(work)) {
        return work.getResult();
      }
      if (attempt == NUM_RETRIES) {
        throw e;  // Give up.
      }
      sleeper.sleepUninterruptibly(Duration.millis(sleepMillis));
      logger.infofmt(e, "Retrying %s, attempt %s", e.getClass().getSimpleName(), attempt);
    }
  }
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
@Override
public final void map(@Nullable final DomainResource domain) {
  if (domain == null) {
    // Emit a single value so that the reducer always runs.  The key and value don't matter.
    emit(true, true);
    return;
  }
  Key<HostResource> referencingHostKey = null;
  for (DnsRefreshRequest request : refreshRequests) {
    if (isActive(domain, request.lastUpdateTime())
        && domain.getNameservers().contains(request.hostKey())) {
      referencingHostKey = request.hostKey();
      break;
    }
  }
  if (referencingHostKey != null) {
    retrier.callWithRetry(
        () -> dnsQueue.addDomainRefreshTask(domain.getFullyQualifiedDomainName()),
        TransientFailureException.class);
    logger.infofmt(
        "Enqueued DNS refresh for domain %s referenced by host %s.",
        domain.getFullyQualifiedDomainName(), referencingHostKey);
    getContext().incrementCounter("domains refreshed");
  } else {
    getContext().incrementCounter("domains not refreshed");
  }

  // Don't catch errors -- we allow the mapreduce to terminate on any errors that can't be
  // resolved by retrying the transaction. The reducer only fires if the mapper completes
  // without errors, meaning that it is acceptable to delete all tasks.
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
/**
 * Deletes a list of tasks associated with deletion requests from the async delete queue using a
 * retrier.
 */
private void deleteStaleTasksWithRetry(final List<DeletionRequest> deletionRequests) {
  if (deletionRequests.isEmpty()) {
    return;
  }
  final List<TaskHandle> tasks =
      deletionRequests.stream().map(DeletionRequest::task).collect(toImmutableList());
  retrier.callWithRetry(() -> queue.deleteTask(tasks), TransientFailureException.class);
  deletionRequests.forEach(
      deletionRequest ->
          asyncFlowMetrics.recordAsyncFlowResult(
              deletionRequest.getMetricOperationType(),
              OperationResult.STALE,
              deletionRequest.requestedTime()));
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
/** Returns handles for a batch of tasks, leased for the specified duration. */
public List<TaskHandle> leaseTasks(Duration leaseDuration) {
  try {
    int numTasks = queue.fetchStatistics().getNumTasks();
    logger.logfmt(
        (numTasks >= leaseTasksBatchSize) ? Level.WARNING : Level.INFO,
        "There are %d tasks in the DNS queue '%s'.",
        numTasks,
        DNS_PULL_QUEUE_NAME);
    return queue.leaseTasks(leaseDuration.getMillis(), MILLISECONDS, leaseTasksBatchSize);
  } catch (TransientFailureException | DeadlineExceededException e) {
    logger.severe(e, "Failed leasing tasks too fast");
    return ImmutableList.of();
  }
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
/** Delete the task, removing it from the queue permanently. */
public void deleteTask(TaskHandle task) {
  try {
    queue.deleteTask(task);
  } catch (TransientFailureException | DeadlineExceededException e) {
    logger.severe(e, "Failed deleting tasks too fast");
  }
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
/** Delete a list of tasks, removing them from the queue permanently. */
public void deleteTasks(List<TaskHandle> tasks) {
  try {
    queue.deleteTask(tasks);
  } catch (TransientFailureException | DeadlineExceededException e) {
    logger.severe(e, "Failed deleting tasks too fast");
  }
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
/**
 * Adds tasks to an App Engine task queue in a reliable manner.
 *
 * <p>This is the same as {@link Queue#add(Iterable)} except it'll automatically retry with
 * exponential backoff if {@link TransientFailureException} is thrown.
 *
 * @throws TransientFailureException if retrying failed for the maximum period of time, or an
 *     {@link InterruptedException} told us to stop trying
 * @return successfully enqueued tasks
 */
public List<TaskHandle> enqueue(final Queue queue, final Iterable<TaskOptions> tasks) {
  return retrier.callWithRetry(
      () -> {
        for (TaskOptions task : tasks) {
          logger.infofmt(
              "Enqueuing queue='%s' endpoint='%s'", queue.getQueueName(), task.getUrl());
        }
        return queue.add(tasks);
      },
      TransientFailureException.class);
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Test
public void test_loadAllTasks_retryLogic_thirdTrysTheCharm() throws Exception {
  Queue queue = mock(Queue.class);
  TaskHandle task = new TaskHandle(TaskOptions.Builder.withTaskName("blah"), "blah");
  when(queue.leaseTasks(any(LeaseOptions.class)))
      .thenThrow(TransientFailureException.class)
      .thenThrow(DeadlineExceededException.class)
      .thenReturn(ImmutableList.of(task), ImmutableList.of());
  assertThat(LordnTask.loadAllTasks(queue, "tld")).containsExactly(task);
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Test
public void test_loadAllTasks_retryLogic_allFailures() throws Exception {
  Queue queue = mock(Queue.class);
  when(queue.leaseTasks(any(LeaseOptions.class))).thenThrow(TransientFailureException.class);
  RuntimeException thrown =
      expectThrows(RuntimeException.class, () -> LordnTask.loadAllTasks(queue, "tld"));
  assertThat(thrown).hasMessageThat().contains("Error leasing tasks");
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
@Test
public void testTransact_transientFailureException_retries() {
  assertThat(ofy().transact(new Work<Integer>() {

    int count = 0;

    @Override
    public Integer run() {
      count++;
      if (count == 3) {
        return count;
      }
      throw new TransientFailureException("");
    }})).isEqualTo(3);
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
@Test
public void testEnqueue_twoTransientErrorsThenSuccess_stillWorksAfterSleeping() throws Exception {
  when(queue.add(ImmutableList.of(task)))
      .thenThrow(new TransientFailureException(""))
      .thenThrow(new TransientFailureException(""))
      .thenReturn(ImmutableList.of(handle));
  assertThat(taskEnqueuer.enqueue(queue, task)).isSameAs(handle);
  verify(queue, times(3)).add(ImmutableList.of(task));
  assertThat(clock.nowUtc()).isEqualTo(DateTime.parse("2000-01-01T00:00:00.6Z"));  // 200 + 400ms
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
@Test
public void testEnqueue_maxRetries_givesUp() throws Exception {
  when(queue.add(ImmutableList.of(task)))
      .thenThrow(new TransientFailureException("one"))
      .thenThrow(new TransientFailureException("two"))
      .thenThrow(new TransientFailureException("three"))
      .thenThrow(new TransientFailureException("four"));
  TransientFailureException thrown =
      expectThrows(TransientFailureException.class, () -> taskEnqueuer.enqueue(queue, task));
  assertThat(thrown).hasMessageThat().contains("three");
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
@Test
public void testEnqueue_transientErrorThenInterrupt_throwsTransientError() throws Exception {
  when(queue.add(ImmutableList.of(task))).thenThrow(new TransientFailureException(""));
  try {
    Thread.currentThread().interrupt();
    assertThrows(TransientFailureException.class, () -> taskEnqueuer.enqueue(queue, task));
  } finally {
    Thread.interrupted();  // Clear interrupt state so it doesn't pwn other tests.
  }
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
@Override
public void export() {
  if (projectId == null) {
    logger.info("Exporting metrics is disabled because no project id was specified");
    return;
  }
  // Export the metrics on a task queue so that we don't block the http response on BigQuery.
  // If this fails, retry twice, after 2 seconds and then 4 seconds, and then give up.
  for (int delaySeconds : ImmutableList.of(0, 2, 4)) {
    sleepUninterruptibly(delaySeconds, SECONDS);
    try {
      getDefaultQueue().add(withUrl("/task/metrics")
          .param("insertid", UUID.randomUUID().toString())
          .param("path", actionPath.endsWith("/*")
              ? actionPath.substring(0, actionPath.length() - 2)
              : actionPath)
          .param("method", method)
          .param("tld", tld)
          .param("starttime", String.valueOf(MILLISECONDS.toSeconds(startTimeMillis)))
          .param("endtime", String.valueOf(MILLISECONDS.toSeconds(System.currentTimeMillis())))
          .param("responsecode", String.valueOf(responseCode))
          .param("activity", Joiner.on(' ').join(activities)));
      return;
    } catch (TransientFailureException e) {
      // Log and swallow.
      logger.log(Level.WARNING, e.getMessage(), e);
    }
    logger.severe("Failed to create metric exporting task.");
  }
}
 

import com.google.appengine.api.taskqueue.TransientFailureException; //导入依赖的package包/类
public static boolean queueWithRetries(String queueName, TaskOptions task, String description) {
  Queue queue = QueueFactory.getQueue(queueName);
  for (int i = 0; i < QUEUE_RETRIES; i++) {
    try {
      queue.add(task);
      return true;
    } catch (TransientFailureException e) {
      LOG.warning("Retrying queue add for task due to queue failure: " + description);
    }
  }
  LOG.severe("Could not enqueue task after " + QUEUE_RETRIES + "retries: " + description);
  return false;
}